Make the LanguageServerProtocolJSONRPC module build with strict concurrency enabled

Author: ahoppen

Package.swift

@@ -119,7 +119,8 @@ let package = Package(
         "LanguageServerProtocol",
         "LSPLogging",
       ],
-      exclude: ["CMakeLists.txt"]
+      exclude: ["CMakeLists.txt"],
+      swiftSettings: [.enableExperimentalFeature("StrictConcurrency")]
     ),
 
     .testTarget(

Sources/LSPLogging/CustomLogStringConvertible.swift

@@ -15,7 +15,7 @@ import Foundation
 
 /// An object that can printed for logging and also offers a redacted description
 /// when logging in contexts in which private information shouldn't be captured.
-public protocol CustomLogStringConvertible: CustomStringConvertible {
+public protocol CustomLogStringConvertible: CustomStringConvertible, Sendable {
   /// A full description of the object.
   var description: String { get }
 
@@ -30,7 +30,7 @@ public protocol CustomLogStringConvertible: CustomStringConvertible {
 /// There currently is no way to get equivalent functionality in pure Swift. We
 /// thus pass this object to OSLog, which just forwards to `description` or
 /// `redactedDescription` of an object that implements `CustomLogStringConvertible`.
-public class CustomLogStringConvertibleWrapper: NSObject {
+public final class CustomLogStringConvertibleWrapper: NSObject, Sendable {
   private let underlyingObject: any CustomLogStringConvertible
 
   fileprivate init(_ underlyingObject: any CustomLogStringConvertible) {

Sources/LanguageServerProtocol/Connection.swift

@@ -21,12 +21,12 @@ public protocol Connection: AnyObject, Sendable {
   /// Send a request and (asynchronously) receive a reply.
   func send<Request: RequestType>(
     _ request: Request,
-    reply: @escaping (LSPResult<Request.Response>) -> Void
+    reply: @escaping @Sendable (LSPResult<Request.Response>) -> Void
   ) -> RequestID
 }
 
 /// An abstract message handler, such as a language server or client.
-public protocol MessageHandler: AnyObject {
+public protocol MessageHandler: AnyObject, Sendable {
 
   /// Handle a notification without a reply.
   ///

Sources/LanguageServerProtocolJSONRPC/JSONRPCConnection.swift

@@ -30,8 +30,8 @@ public final class JSONRPCConnection: Connection {
   /// The message handler that handles requests and notifications sent through this connection.
   ///
   /// Access to this must be be guaranteed to be sequential to avoid data races. Currently, all access are
-  ///  - `init`: Can trivially only be called once
-  ///  - `start`: Is required to be call in the same sequential code region as the initializer, so
+  ///  - `init`: Reference to `JSONRPCConnection` trivially can't have escaped to other isolation domains yet.
+  ///  - `start`: Is required to be call in the same serial code region as the initializer, so
   ///    `JSONRPCConnection` can't have escaped to other isolation domains yet.
   ///  - `deinit`: Can also only trivially be called once.
   nonisolated(unsafe) private var receiveHandler: MessageHandler?
@@ -56,35 +56,54 @@ public final class JSONRPCConnection: Connection {
   /// Current state of the connection, used to ensure correct usage.
   ///
   /// Access to this must be be guaranteed to be sequential to avoid data races. Currently, all access are
-  ///  - `init`: Can trivially only be called once
-  ///  - `start`: Is required to be called in the same sequential region as the initializer, so
+  ///  - `init`: Reference to `JSONRPCConnection` trivially can't have escaped to other isolation domains yet.
+  ///  - `start`: Is required to be called in the same serial region as the initializer, so
   ///    `JSONRPCConnection` can't have escaped to other isolation domains yet.
   ///  - `_close`: Synchronized on `queue`.
   ///  - `readyToSend`: Synchronized on `queue`.
   ///  - `deinit`: Can also only trivially be called once.
-  nonisolated(unsafe) private var state: State
+  private nonisolated(unsafe) var state: State
 
   /// Buffer of received bytes that haven't been parsed.
+  ///
+  /// Access to this must be be guaranteed to be sequential to avoid data races. Currently, all access are
+  ///  - The `receiveIO` handler: This is synchronized on `queue`.
+  ///  - `requestBufferIsEmpty`: Also synchronized on `queue`.
+  private nonisolated(unsafe) var requestBuffer: [UInt8] = []
+
   @_spi(Testing)
-  public var _requestBuffer: [UInt8] = []
+  public var requestBufferIsEmpty: Bool {
+    queue.sync {
+      requestBuffer.isEmpty
+    }
+  }
 
-  private var _nextRequestID: Int = 0
+  /// An integer that hasn't been used for a request ID yet.
+  ///
+  /// Access to this must be be guaranteed to be sequential to avoid data races. Currently, all access are
+  ///  - `nextRequestID()`: This is synchronized on `queue`.
+  private nonisolated(unsafe) var nextRequestIDStorage: Int = 0
 
-  struct OutstandingRequest {
+  struct OutstandingRequest: Sendable {
     var responseType: ResponseType.Type
-    var replyHandler: (LSPResult<Any>) -> Void
+    var replyHandler: @Sendable (LSPResult<Any>) -> Void
   }
 
-  /// The set of currently outstanding outgoing requests along with information about how to decode and handle their responses.
-  private var outstandingRequests: [RequestID: OutstandingRequest] = [:]
+  /// The set of currently outstanding outgoing requests along with information about how to decode and handle their
+  /// responses.
+  ///
+  /// All accesses to `outstandingRequests` must be on `queue` to avoid race conditions.
+  private nonisolated(unsafe) var outstandingRequests: [RequestID: OutstandingRequest] = [:]
 
   /// A handler that will be called asynchronously when the connection is being
   /// closed.
-  private var closeHandler: (() async -> Void)? = nil
+  ///
+  /// There are no race conditions to `closeHandler` because it is only set from `start`, which is required to be called
+  /// in the same serial code region domain as the initializer, so it's serial and the `JSONRPCConnection` can't
+  /// have escaped to other isolation domains yet.
+  private nonisolated(unsafe) var closeHandler: (@Sendable () async -> Void)? = nil
 
-  /// - Important: `start` must be called within the same sequential code region that created the `JSONRPCConnection`.
-  ///   This means that no `await` calls must exist between the creation of the connection and the call to `start` and
-  ///   the `JSONRPCConnection` must not be accessible to any other threads yet.
+  /// - Important: `start` must be called before sending any data over the `JSONRPCConnection`.
   public init(
     name: String,
     protocol messageRegistry: MessageRegistry,
@@ -95,7 +114,8 @@ public final class JSONRPCConnection: Connection {
     self.name = name
     self.receiveHandler = nil
     #if os(Linux) || os(Android)
-    // We receive a `SIGPIPE` if we write to a pipe that points to a crashed process. This in particular happens if the target of a `JSONRPCConnection` has crashed and we try to send it a message.
+    // We receive a `SIGPIPE` if we write to a pipe that points to a crashed process. This in particular happens if the
+    // target of a `JSONRPCConnection` has crashed and we try to send it a message.
     // On Darwin, `DispatchIO` ignores `SIGPIPE` for the pipes handled by it, but that features is not available on Linux.
     // Instead, globally ignore `SIGPIPE` on Linux to prevent us from crashing if the `JSONRPCConnection`'s target crashes.
     globallyDisableSigpipe()
@@ -168,49 +188,49 @@ public final class JSONRPCConnection: Connection {
   ///
   /// - parameter receiveHandler: The message handler to invoke for requests received on the `inFD`.
   ///
-  /// - Important: `start` must be called within the same sequential code region that created the `JSONRPCConnection`.
-  ///   This means that no `await` calls must exist between the creation of the connection and the call to `start` and
-  ///   the `JSONRPCConnection` must not be accessible to any other threads yet.
-  public func start(receiveHandler: MessageHandler, closeHandler: @escaping () async -> Void = {}) {
-    precondition(state == .created)
-    state = .running
-    self.receiveHandler = receiveHandler
-    self.closeHandler = closeHandler
-
-    receiveIO.read(offset: 0, length: Int.max, queue: queue) { done, data, errorCode in
-      guard errorCode == 0 else {
-        #if !os(Windows)
-        if errorCode != POSIXError.ECANCELED.rawValue {
-          logger.error("IO error reading \(errorCode)")
+  /// - Important: `start` must be called before sending any data over the `JSONRPCConnection`.
+  public func start(receiveHandler: MessageHandler, closeHandler: @escaping @Sendable () async -> Void = {}) {
+    queue.sync {
+      precondition(state == .created)
+      state = .running
+      self.receiveHandler = receiveHandler
+      self.closeHandler = closeHandler
+
+      receiveIO.read(offset: 0, length: Int.max, queue: queue) { done, data, errorCode in
+        guard errorCode == 0 else {
+          #if !os(Windows)
+          if errorCode != POSIXError.ECANCELED.rawValue {
+            logger.error("IO error reading \(errorCode)")
+          }
+          #endif
+          if done { self.closeAssumingOnQueue() }
+          return
         }
-        #endif
-        if done { self.closeOnQueue() }
-        return
-      }
-
-      if done {
-        self.closeOnQueue()
-        return
-      }
 
-      guard let data = data, !data.isEmpty else {
-        return
-      }
+        if done {
+          self.closeAssumingOnQueue()
+          return
+        }
 
-      // Parse and handle any messages in `buffer + data`, leaving any remaining unparsed bytes in `buffer`.
-      if self._requestBuffer.isEmpty {
-        data.withUnsafeBytes { (pointer: UnsafePointer<UInt8>) in
-          let rest = self.parseAndHandleMessages(from: UnsafeBufferPointer(start: pointer, count: data.count))
-          self._requestBuffer.append(contentsOf: rest)
+        guard let data = data, !data.isEmpty else {
+          return
         }
-      } else {
-        self._requestBuffer.append(contentsOf: data)
-        var unused = 0
-        self._requestBuffer.withUnsafeBufferPointer { buffer in
-          let rest = self.parseAndHandleMessages(from: buffer)
-          unused = rest.count
+
+        // Parse and handle any messages in `buffer + data`, leaving any remaining unparsed bytes in `buffer`.
+        if self.requestBuffer.isEmpty {
+          data.withUnsafeBytes { (pointer: UnsafePointer<UInt8>) in
+            let rest = self.parseAndHandleMessages(from: UnsafeBufferPointer(start: pointer, count: data.count))
+            self.requestBuffer.append(contentsOf: rest)
+          }
+        } else {
+          self.requestBuffer.append(contentsOf: data)
+          var unused = 0
+          self.requestBuffer.withUnsafeBufferPointer { buffer in
+            let rest = self.parseAndHandleMessages(from: buffer)
+            unused = rest.count
+          }
+          self.requestBuffer.removeFirst(self.requestBuffer.count - unused)
         }
-        self._requestBuffer.removeFirst(self._requestBuffer.count - unused)
       }
     }
   }
@@ -231,7 +251,10 @@ public final class JSONRPCConnection: Connection {
   }
 
   /// Parse and handle all messages in `bytes`, returning a slice containing any remaining incomplete data.
+  ///
+  /// - Important: Must be called on `queue`
   func parseAndHandleMessages(from bytes: UnsafeBufferPointer<UInt8>) -> UnsafeBufferPointer<UInt8>.SubSequence {
+    dispatchPrecondition(condition: .onQueue(queue))
     let decoder = JSONDecoder()
 
     // Set message registry to use for model decoding.
@@ -299,7 +322,10 @@ public final class JSONRPCConnection: Connection {
   }
 
   /// Handle a single message by dispatching it to `receiveHandler` or an appropriate reply handler.
+  ///
+  /// - Important: Must be called on `queue`
   func handle(_ message: JSONRPCMessage) {
+    dispatchPrecondition(condition: .onQueue(queue))
     switch message {
     case .notification(let notification):
       notification._handle(self.receiveHandler!)
@@ -341,11 +367,11 @@ public final class JSONRPCConnection: Connection {
     sendIO.write(offset: 0, data: dispatchData, queue: sendQueue) { [weak self] done, _, errorCode in
       if errorCode != 0 {
         logger.error("IO error sending message \(errorCode)")
-        if done {
+        if done, let self {
           // An unrecoverable error occurs on the channel’s file descriptor.
           // Close the connection.
-          self?.queue.async {
-            self?.closeOnQueue()
+          self.queue.async {
+            self.closeAssumingOnQueue()
           }
         }
       }
@@ -398,13 +424,13 @@ public final class JSONRPCConnection: Connection {
   /// The user-provided close handler will be called *asynchronously* when all outstanding I/O
   /// operations have completed. No new I/O will be accepted after `close` returns.
   public func close() {
-    queue.sync { closeOnQueue() }
+    queue.sync { closeAssumingOnQueue() }
   }
 
   /// Close the connection, assuming that the code is already executing on `queue`.
   ///
   /// - Important: Must be called on `queue`.
-  func closeOnQueue() {
+  private func closeAssumingOnQueue() {
     dispatchPrecondition(condition: .onQueue(queue))
     sendQueue.sync {
       guard state == .running else { return }
@@ -419,9 +445,12 @@ public final class JSONRPCConnection: Connection {
   }
 
   /// Request id for the next outgoing request.
-  func nextRequestID() -> RequestID {
-    _nextRequestID += 1
-    return .number(_nextRequestID)
+  ///
+  /// - Important: Must be called on `queue`
+  private func nextRequestID() -> RequestID {
+    dispatchPrecondition(condition: .onQueue(queue))
+    nextRequestIDStorage += 1
+    return .number(nextRequestIDStorage)
   }
 
   // MARK: Connection interface
@@ -444,7 +473,7 @@ public final class JSONRPCConnection: Connection {
   /// When the receiving end replies to the request, execute `reply` with the response.
   public func send<Request: RequestType>(
     _ request: Request,
-    reply: @escaping (LSPResult<Request.Response>) -> Void
+    reply: @escaping @Sendable (LSPResult<Request.Response>) -> Void
   ) -> RequestID {
     let id: RequestID = self.queue.sync {
       let id = nextRequestID()

Sources/SourceKitLSP/CapabilityRegistry.swift

@@ -196,7 +196,7 @@ public final actor CapabilityRegistry {
   public func registerDidChangeWatchedFiles(
     watchers: [FileSystemWatcher],
     server: SourceKitServer
-  ) {
+  ) async {
     guard clientHasDynamicDidChangeWatchedFilesRegistration else { return }
     if let registration = didChangeWatchedFiles {
       if watchers != registration.watchers {
@@ -216,11 +216,11 @@ public final actor CapabilityRegistry {
 
     self.didChangeWatchedFiles = registrationOptions
 
-    let _ = server.client.send(RegisterCapabilityRequest(registrations: [registration])) { result in
-      if let error = result.failure {
-        logger.error("Failed to dynamically register for watched files: \(error.forLogging)")
-        self.didChangeWatchedFiles = nil
-      }
+    do {
+      _ = try await server.client.send(RegisterCapabilityRequest(registrations: [registration]))
+    } catch {
+      logger.error("Failed to dynamically register for watched files: \(error.forLogging)")
+      self.didChangeWatchedFiles = nil
     }
   }
 

Sources/SourceKitLSP/Clang/ClangLanguageServer.swift

@@ -445,7 +445,7 @@ extension ClangLanguageServerShim {
     await withCheckedContinuation { continuation in
       _ = clangd.send(ShutdownRequest()) { _ in
         Task {
-          self.clangd.send(ExitNotification())
+          await self.clangd.send(ExitNotification())
           continuation.resume()
         }
       }

Sources/SourceKitLSP/SourceKitServer.swift

@@ -138,24 +138,33 @@ final actor WorkDoneProgressState {
       state = .creating
       // Discard the handle. We don't support cancellation of the creation of a work done progress.
       _ = server.client.send(CreateWorkDoneProgressRequest(token: token)) { result in
-        if result.success != nil {
-          if self.activeTasks == 0 {
-            // ActiveTasks might have been decreased while we created the `WorkDoneProgress`
-            self.state = .noProgress
-            server.client.send(WorkDoneProgress(token: self.token, value: .end(WorkDoneProgressEnd())))
-          } else {
-            self.state = .created
-            server.client.send(
-              WorkDoneProgress(token: self.token, value: .begin(WorkDoneProgressBegin(title: self.title)))
-            )
-          }
-        } else {
-          self.state = .progressCreationFailed
+        Task {
+          await self.handleCreateWorkDoneProgressResponse(result, server: server)
         }
       }
     }
   }
 
+  private func handleCreateWorkDoneProgressResponse(
+    _ result: Result<VoidResponse, ResponseError>,
+    server: SourceKitServer
+  ) {
+    if result.success != nil {
+      if self.activeTasks == 0 {
+        // ActiveTasks might have been decreased while we created the `WorkDoneProgress`
+        self.state = .noProgress
+        server.client.send(WorkDoneProgress(token: self.token, value: .end(WorkDoneProgressEnd())))
+      } else {
+        self.state = .created
+        server.client.send(
+          WorkDoneProgress(token: self.token, value: .begin(WorkDoneProgressBegin(title: self.title)))
+        )
+      }
+    } else {
+      self.state = .progressCreationFailed
+    }
+  }
+
   /// End a new task stated using `startProgress`.
   ///
   /// If this drops the active task count to 0, the work done progress is ended on the client.

Tests/LanguageServerProtocolJSONRPCTests/ConnectionTests.swift

@@ -90,9 +90,9 @@ class ConnectionTests: XCTestCase {
 
     try await fulfillmentOfOrThrow([expectation2])
 
-    // Close the connection before accessing _requestBuffer, which ensures we don't race.
+    // Close the connection before accessing requestBuffer, which ensures we don't race.
     connection.serverToClientConnection.close()
-    XCTAssertEqual(connection.serverToClientConnection._requestBuffer, [])
+    XCTAssert(connection.serverToClientConnection.requestBufferIsEmpty)
   }
 
   func testEchoError() {