fix: avoid leaking CheckedContinuations by removing possible Channel actor reentrancy

In the Channel.fulfill and Channel.fail methods, the line `await state.removeAllWaiters()` allows for actor reentrance. This means that after resuming all the CheckedContinuations (waiters), but before removing them, another waiter could be added to the array. Then once the `state.removeAllWaiters` resumes execution, any CheckedContinuations would be removed without having been resumed. This leads to leaking CheckedContinuations.

This was observed in my application, where I would get a logger message

SWIFT TASK CONTINUATION MISUSE: value leaked its continuation without resuming it. This may cause tasks waiting on it to remain suspended forever.

which originates from Swift here: https://github.com/swiftlang/swift/blob/b06eed151c8aa2bc2f4e081b0bb7b5e5c65f3bba/stdlib/public/Concurrency/CheckedContinuation.swift#L82

I verified that Channel.value was the culprit by renaming it and observing the changed name in the CONTINUATION MISUSE message.

This change set removes the possibility of actor reentrancy and leaking CheckedContinuations. By inlining the State data members in Channel and deleting the State actor completely, we remove all await calls inside Channel.fulfill and Channel.fail.

Author: Brandon Schoenfeld

Sources/AsyncDataLoader/Channel/Channel.swift renamed to Sources/AsyncDataLoader/Channel.swift

@@ -1,18 +1,22 @@
 actor Channel<Success: Sendable, Failure: Error>: Sendable {
-    private var state = State<Success, Failure>()
+    private var waiters = [Waiter<Success, Failure>]()
+    private var result: Success?
+    private var failure: Failure?
 }
 
+typealias Waiter<Success, Failure> = CheckedContinuation<Success, Error>
+
 extension Channel {
     @discardableResult
-    func fulfill(_ value: Success) async -> Bool {
-        if await state.result == nil {
-            await state.setResult(result: value)
+    func fulfill(_ value: Success) -> Bool {
+        if result == nil {
+            result = value
 
-            for waiters in await state.waiters {
+            for waiters in waiters {
                 waiters.resume(returning: value)
             }
 
-            await state.removeAllWaiters()
+            waiters.removeAll()
 
             return false
         }
@@ -21,15 +25,15 @@ extension Channel {
     }
 
     @discardableResult
-    func fail(_ failure: Failure) async -> Bool {
-        if await state.failure == nil {
-            await state.setFailure(failure: failure)
+    func fail(_ failure: Failure) -> Bool {
+        if self.failure == nil {
+            self.failure = failure
 
-            for waiters in await state.waiters {
+            for waiters in waiters {
                 waiters.resume(throwing: failure)
             }
 
-            await state.removeAllWaiters()
+            waiters.removeAll()
 
             return false
         }
@@ -41,12 +45,12 @@ extension Channel {
         get async throws {
             try await withCheckedThrowingContinuation { continuation in
                 Task {
-                    if let result = await state.result {
+                    if let result = self.result {
                         continuation.resume(returning: result)
-                    } else if let failure = await self.state.failure {
+                    } else if let failure = self.failure {
                         continuation.resume(throwing: failure)
                     } else {
-                        await state.appendWaiters(waiters: continuation)
+                        waiters.append(continuation)
                     }
                 }
             }