Synchronously send error response on message decoding failure (#334)

* Replaces two logs since those don't result in response messages anymore
* Adds a func for sending messages synchronously, to get the two messages out before fatalError

Author: flowtoolz

Sources/LanguageServerProtocolJSONRPC/JSONRPCConnection.swift

@@ -172,6 +172,13 @@ public final class JSONRPCConnection {
     }
     return ready
   }
+  
+  /// *Public for testing*
+  public func _send(_ message: JSONRPCMessage, async: Bool = true) {
+    send(async: async) { encoder in
+      try encoder.encode(message)
+    }
+  }
 
   /// Parse and handle all messages in `bytes`, returning a slice containing any remaining incomplete data.
   func parseAndHandleMessages(from bytes: UnsafeBufferPointer<UInt8>) -> UnsafeBufferPointer<UInt8>.SubSequence {
@@ -209,9 +216,7 @@ public final class JSONRPCConnection {
         switch error.messageKind {
           case .request:
             if let id = error.id {
-              send { encoder in
-                try encoder.encode(JSONRPCMessage.errorResponse(ResponseError(error), id: id))
-              }
+              _send(.errorResponse(ResponseError(error), id: id))
               continue MESSAGE_LOOP
             }
           case .response:
@@ -229,17 +234,19 @@ public final class JSONRPCConnection {
               continue MESSAGE_LOOP
             }
           case .unknown:
-            log("error decoding message: \(error.message)", level: .error)
+            _send(.errorResponse(ResponseError(error), id: nil),
+                 async: false) // synchronous because the following fatalError
             break
         }
         // FIXME: graceful shutdown?
-        Logger.shared.flush()
         fatalError("fatal error encountered decoding message \(error)")
 
       } catch {
-        log("error decoding message: \(error.localizedDescription)", level: .error)
+        let responseError = ResponseError(code: .parseError,
+                                          message: "Failed to decode message. \(error.localizedDescription)")
+        _send(.errorResponse(responseError, id: nil),
+             async: false) // synchronous because the following fatalError
         // FIXME: graceful shutdown?
-        Logger.shared.flush()
         fatalError("fatal error encountered decoding message \(error)")
       }
     }
@@ -265,16 +272,21 @@ public final class JSONRPCConnection {
       }
       outstanding.replyHandler(.success(response))
     case .errorResponse(let error, id: let id):
+      guard let id = id else {
+        log("Received error response for unknown request: \(error.message)", level: .error)
+        return
+      }
       guard let outstanding = outstandingRequests.removeValue(forKey: id) else {
-        log("Unknown request for \(id)", level: .error)
+        log("No outstanding requests for request ID \(id)", level: .error)
         return
       }
       outstanding.replyHandler(.failure(error))
     }
   }
 
   /// *Public for testing*.
-  public func send(_rawData dispatchData: DispatchData) {
+  public func send(_rawData dispatchData: DispatchData,
+                   handleCompletion: (() -> Void)? = nil) {
     guard readyToSend() else { return }
 
     sendIO.write(offset: 0, data: dispatchData, queue: sendQueue) { [weak self] done, _, errorCode in
@@ -283,13 +295,16 @@ public final class JSONRPCConnection {
         if done {
           self?.queue.async {
             self?._close()
+            handleCompletion?()
           }
         }
+      } else if done {
+        handleCompletion?()
       }
     }
   }
 
-  func send(messageData: Data) {
+  func send(messageData: Data, handleCompletion: (() -> Void)? = nil) {
 
     var dispatchData = DispatchData.empty
     let header = "Content-Length: \(messageData.count)\r\n\r\n"
@@ -300,10 +315,22 @@ public final class JSONRPCConnection {
       dispatchData.append(rawBufferPointer)
     }
 
-    send(_rawData: dispatchData)
+    send(_rawData: dispatchData, handleCompletion: handleCompletion)
   }
 
-  func send(encoding: (JSONEncoder) throws -> Data) {
+  private func sendMessageSynchronously(_ messageData: Data,
+                                        timeoutInSeconds seconds: Int) {
+    let synchronizationSemaphore = DispatchSemaphore(value: 0)
+    
+    send(messageData: messageData) {
+        synchronizationSemaphore.signal()
+    }
+    
+    // blocks until timeout expires or message sending completes
+    _ = synchronizationSemaphore.wait(timeout: .now() + .seconds(seconds))
+  }
+  
+  func send(async: Bool = true, encoding: (JSONEncoder) throws -> Data) {
     guard readyToSend() else { return }
 
     let encoder = JSONEncoder()
@@ -317,7 +344,11 @@ public final class JSONRPCConnection {
       fatalError("unexpected error while encoding response: \(error)")
     }
 
-    send(messageData: data)
+    if async {
+      send(messageData: data)
+    } else {
+      sendMessageSynchronously(data, timeoutInSeconds: 3)
+    }
   }
 
   /// Close the connection.

Sources/LanguageServerProtocolJSONRPC/MessageCoding.swift

@@ -17,7 +17,7 @@ public enum JSONRPCMessage {
   case notification(NotificationType)
   case request(_RequestType, id: RequestID)
   case response(ResponseType, id: RequestID)
-  case errorResponse(ResponseError, id: RequestID)
+  case errorResponse(ResponseError, id: RequestID?)
 }
 
 extension CodingUserInfoKey {
@@ -96,7 +96,7 @@ extension JSONRPCMessage: Codable {
 
         self = .response(result, id: id)
 
-      case (let id?, nil, _, let error?):
+      case (let id, nil, _, let error?):
         msgKind = .response
         self = .errorResponse(error, id: id)
 

Tests/LanguageServerProtocolJSONRPCTests/CodingTests.swift

@@ -126,6 +126,17 @@ final class CodingTests: XCTestCase {
       "jsonrpc" : "2.0"
     }
     """)
+    
+    checkMessageCoding(ResponseError.cancelled, id: nil, json: """
+    {
+      "error" : {
+        "code" : -32800,
+        "message" : "request cancelled"
+      },
+      "id" : null,
+      "jsonrpc" : "2.0"
+    }
+    """)
   }
 
   func testMessageDecodingError() {
@@ -147,10 +158,6 @@ final class CodingTests: XCTestCase {
     {"jsonrpc":"2.0","result":{}}
     """)
 
-    checkMessageDecodingError(MessageDecodingError.invalidRequest("message not recognized as request, response or notification"), json: """
-    {"jsonrpc":"2.0","error":{"code":-32000,"message":""}}
-    """)
-
     checkMessageDecodingError(MessageDecodingError.methodNotFound("unknown", messageKind: .notification), json: """
     {"jsonrpc":"2.0","method":"unknown"}
     """)
@@ -255,7 +262,7 @@ private func checkMessageCoding<Response>(_ value: Response, id: RequestID, json
   }
 }
 
-private func checkMessageCoding(_ value: ResponseError, id: RequestID, json: String, file: StaticString = #file, line: UInt = #line) {
+private func checkMessageCoding(_ value: ResponseError, id: RequestID?, json: String, file: StaticString = #file, line: UInt = #line) {
   checkCoding(JSONRPCMessage.errorResponse(value, id: id), json: json, userInfo: defaultCodingInfo, file: file, line: line) {
 
     guard case JSONRPCMessage.errorResponse(let decodedValue, let decodedID) = $0 else {

Tests/LanguageServerProtocolJSONRPCTests/ConnectionTests.swift

@@ -211,6 +211,20 @@ class ConnectionTests: XCTestCase {
 
     waitForExpectations(timeout: 10)
   }
+  
+  func testSendSynchronouslyBeforeClose() {
+    let client = connection.client
+
+    let expectation = self.expectation(description: "received notification")
+    client.handleNextNotification { (note: Notification<EchoNotification>) in
+      expectation.fulfill()
+    }
+    let notification = EchoNotification(string: "about to close!")
+    connection.serverConnection._send(.notification(notification), async: false)
+    connection.serverConnection.close()
+
+    waitForExpectations(timeout: 10)
+  }
 
   /// We can explicitly close a connection, but the connection also
   /// automatically closes itself if the pipe is closed (or has an error).

Tests/LanguageServerProtocolJSONRPCTests/XCTestManifests.swift

@@ -35,6 +35,7 @@ extension ConnectionTests {
         ("testRound", testRound),
         ("testSendAfterClose", testSendAfterClose),
         ("testSendBeforeClose", testSendBeforeClose),
+        ("testSendSynchronouslyBeforeClose", testSendSynchronouslyBeforeClose),
         ("testUnexpectedResponse", testUnexpectedResponse),
         ("testUnknownNotification", testUnknownNotification),
         ("testUnknownRequest", testUnknownRequest),