Prevent duplicate resume continuation on unary request (#340)

Issue #333 reported an issue where the continuation in the
`UnaryAsyncWrapper` was getting resumed twice, which results in a fatal
error. Additional details are available in the issue, but the gist is -

When the client is configured to use the gRPC protocol (therefore using
swift-nio for networking instead of URLSession), and it is configured
with a short timeout internal, when the server responds in error at
around the same time that the timeout occurs, the callback that resumes
the continuation may fire twice.

I was able to replicate this issue by adjusting the Eliza app's
`ProtocolClient` to have a timeout of 0.25s and to point Eliza at a
different Connect server (my own team's), such that requests to the
Eliza RPCs would result in 404 errors.

This change simply introduces a thread-safe `Bool` to track whether the
continuation has been resumed, and if the callback attempts to fire
twice, it will throw an `assertionFailure` and return instead of calling
`resume` again and causing a fatal error.

Closes #333

---------

Signed-off-by: Eddie Seay <[email protected]>

Author: eseay

.github/workflows/ci.yaml

@@ -123,7 +123,7 @@ jobs:
   run-swiftlint:
     runs-on: ubuntu-latest
     container:
-      image: ghcr.io/realm/swiftlint:0.57.1
+      image: ghcr.io/realm/swiftlint:0.58.2
     steps:
       - uses: actions/checkout@v4
       - name: Run SwiftLint

Libraries/Connect/Internal/Unary/UnaryAsyncWrapper.swift

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+import os.log
 import SwiftProtobuf
 
 /// Internal actor used to wrap closure-based unary API calls in a way that allows them
@@ -21,7 +22,7 @@ import SwiftProtobuf
 /// https://forums.swift.org/t/how-to-use-withtaskcancellationhandler-properly/54341/37
 /// https://stackoverflow.com/q/71898080
 @available(iOS 13, *)
-actor UnaryAsyncWrapper<Output: ProtobufMessage>: Sendable {
+actor UnaryAsyncWrapper<Output: ProtobufMessage> {
     private var cancelable: Cancelable?
     private let sendUnary: PerformClosure
 
@@ -41,20 +42,45 @@ actor UnaryAsyncWrapper<Output: ProtobufMessage>: Sendable {
     ///
     /// - returns: The response/result of the request.
     func send() async -> ResponseMessage<Output> {
-        return await withTaskCancellationHandler(operation: {
-            return await withCheckedContinuation { continuation in
-                if Task.isCancelled {
+        await withTaskCancellationHandler {
+            await withCheckedContinuation { continuation in
+                guard !Task.isCancelled else {
                     continuation.resume(
-                        returning: .init(code: .canceled, result: .failure(.canceled()))
+                        returning: ResponseMessage(
+                            code: .canceled,
+                            result: .failure(.canceled())
+                        )
                     )
-                } else {
-                    self.cancelable = self.sendUnary { response in
-                        continuation.resume(returning: response)
+                    return
+                }
+
+                let hasResumed = Locked(false)
+                self.cancelable = self.sendUnary { response in
+                    // In some circumstances where a request timeout and a server
+                    // error occur at nearly the same moment, the underlying
+                    // `swift-nio` system will trigger this callback twice. This check
+                    // discards the second occurrence to avoid resuming `continuation`
+                    // multiple times, which would result in a crash.
+                    guard !hasResumed.value else {
+                        os_log(
+                            .fault,
+                            """
+                            `sendUnary` received duplicate callback and \
+                            attempted to resume its continuation twice.
+                            """
+                        )
+                        return
                     }
+                    continuation.resume(returning: response)
+                    hasResumed.perform(action: { $0 = true })
                 }
             }
-        }, onCancel: {
-            Task { await self.cancelable?.cancel() }
-        })
+        } onCancel: {
+            // When `Task.cancel` signals for this function to be canceled,
+            // the underlying function will be canceled as well.
+            Task(priority: .high) {
+                await self.cancelable?.cancel()
+            }
+        }
     }
 }