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.swift

@@ -0,0 +1,59 @@
+actor Channel<Success: Sendable, Failure: Error>: Sendable {
+    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) -> Bool {
+        if result == nil {
+            result = value
+
+            for waiter in waiters {
+                waiter.resume(returning: value)
+            }
+
+            waiters.removeAll()
+
+            return false
+        }
+
+        return true
+    }
+
+    @discardableResult
+    func fail(_ failure: Failure) -> Bool {
+        if self.failure == nil {
+            self.failure = failure
+
+            for waiter in waiters {
+                waiter.resume(throwing: failure)
+            }
+
+            waiters.removeAll()
+
+            return false
+        }
+
+        return true
+    }
+
+    var value: Success {
+        get async throws {
+            try await withCheckedThrowingContinuation { continuation in
+                Task {
+                    if let result = self.result {
+                        continuation.resume(returning: result)
+                    } else if let failure = self.failure {
+                        continuation.resume(throwing: failure)
+                    } else {
+                        waiters.append(continuation)
+                    }
+                }
+            }
+        }
+    }
+}