Fun

FranzBusch · FranzBusch · commit 5a4946b3d963 · 2025-08-14T14:12:32.000+02:00
diff --git a/Sources/AsyncAlgorithms/CombineLatest/AsyncCombineLatestManySequence.swift b/Sources/AsyncAlgorithms/CombineLatest/AsyncCombineLatestManySequence.swift
@@ -21,21 +21,18 @@
 /// ``combineLatestMany(_:)`` throws when one of the bases throws. If one of the bases threw any buffered and not yet consumed
 /// values will be dropped.
 @available(AsyncAlgorithms 1.1, *)
-public func combineLatestMany<Element: Sendable>(_ bases: [any CombineLatestManyBase<Element>]) -> AsyncCombineLatestManySequence<Element>
-{
-  AsyncCombineLatestManySequence(bases)
+public func combineLatestMany<Element: Sendable, Failure: Error>(
+    _ bases: [any (AsyncSequence<Element, Failure> & Sendable)]
+) -> some AsyncSequence<[Element], Failure> & Sendable {
+  AsyncCombineLatestManySequence<Element, Failure>(bases)
 }
 
-// TODO: Can we get rid of this typealias?
-@available(AsyncAlgorithms 1.1, *)
-public typealias CombineLatestManyBase<Element: Sendable> = AsyncSequence<Element, any Error> & Sendable
-
 /// An `AsyncSequence` that combines the latest values produced from many asynchronous sequences into an asynchronous sequence of tuples.
 @available(AsyncAlgorithms 1.1, *)
-public struct AsyncCombineLatestManySequence<Element: Sendable>: AsyncSequence, Sendable {
+public struct AsyncCombineLatestManySequence<Element: Sendable, Failure: Error>: AsyncSequence, Sendable {
   public typealias AsyncIterator = Iterator
     
-  typealias Base = AsyncSequence<Element, any Error> & Sendable
+  typealias Base = AsyncSequence<Element, Failure> & Sendable
   let bases: [any Base]
 
   init(_ bases: [any Base]) {
@@ -50,33 +47,34 @@ public struct AsyncCombineLatestManySequence<Element: Sendable>: AsyncSequence,
 
   public struct Iterator: AsyncIteratorProtocol {
     final class InternalClass {
-      private let storage: CombineLatestManyStorage<Element>
+      private let storage: CombineLatestManyStorage<Element, Failure>
 
-      fileprivate init(storage: CombineLatestManyStorage<Element>) {
+      fileprivate init(storage: CombineLatestManyStorage<Element, Failure>) {
         self.storage = storage
       }
 
       deinit {
         self.storage.iteratorDeinitialized()
       }
 
-      func next() async throws -> [Element]? {
-        guard let element = try await self.storage.next() else {
-          return nil
-        }
-
-        // This force unwrap is safe since there must be a third element.
-        return element
+      func next() async throws(Failure) -> [Element]? {
+          fatalError()
+//        guard let element = try await self.storage.next() else {
+//          return nil
+//        }
+//
+//        // This force unwrap is safe since there must be a third element.
+//        return element
       }
     }
 
     let internalClass: InternalClass
 
-    fileprivate init(storage: CombineLatestManyStorage<Element>) {
+    fileprivate init(storage: CombineLatestManyStorage<Element, Failure>) {
       self.internalClass = InternalClass(storage: storage)
     }
 
-    public mutating func next() async throws -> [Element]? {
+    public mutating func next() async throws(Failure) -> [Element]? {
       try await self.internalClass.next()
     }
   }
diff --git a/Sources/AsyncAlgorithms/CombineLatest/CombineLatestManyStateMachine.swift b/Sources/AsyncAlgorithms/CombineLatest/CombineLatestManyStateMachine.swift
@@ -13,17 +13,17 @@ import DequeModule
 
 /// State machine for combine latest
 @available(AsyncAlgorithms 1.1, *)
-struct CombineLatestManyStateMachine<Element: Sendable>: Sendable {
+struct CombineLatestManyStateMachine<Element: Sendable, Failure: Error>: Sendable {
   typealias DownstreamContinuation = UnsafeContinuation<
-    Result<[Element]?, Error>, Never
+    Result<[Element]?, Failure>, Never
   >
-  typealias Base = AsyncSequence<Element, any Error> & Sendable
+  typealias Base = AsyncSequence<Element, Failure> & Sendable
 
   private enum State: Sendable {
     /// Small wrapper for the state of an upstream sequence.
     struct Upstream: Sendable {
       /// The upstream continuation.
-      var continuation: UnsafeContinuation<Void, Error>?
+      var continuation: UnsafeContinuation<Void, Never>?
       /// The produced upstream element.
       var element: Element?
       /// Indicates wether the upstream finished/threw already
@@ -53,7 +53,7 @@ struct CombineLatestManyStateMachine<Element: Sendable>: Sendable {
     )
 
     case upstreamThrew(
-      error: Error
+      error: Failure
     )
 
     /// The state once the downstream consumer stopped, i.e. by dropping all references
@@ -77,10 +77,10 @@ struct CombineLatestManyStateMachine<Element: Sendable>: Sendable {
   /// Actions returned by `iteratorDeinitialized()`.
   enum IteratorDeinitializedAction {
     /// Indicates that the `Task` needs to be cancelled and
-    /// the upstream continuations need to be resumed with a `CancellationError`.
+    /// the upstream continuations need to be resumed with a `CancellationFailure`.
     case cancelTaskAndUpstreamContinuations(
       task: Task<Void, Never>,
-      upstreamContinuations: [UnsafeContinuation<Void, Error>]
+      upstreamContinuations: [UnsafeContinuation<Void, Never>]
     )
   }
 
@@ -151,18 +151,13 @@ struct CombineLatestManyStateMachine<Element: Sendable>: Sendable {
   enum ChildTaskSuspendedAction {
     /// Indicates that the continuation should be resumed which will lead to calling `next` on the upstream.
     case resumeContinuation(
-      upstreamContinuation: UnsafeContinuation<Void, Error>
-    )
-    /// Indicates that the continuation should be resumed with an Error because another upstream sequence threw.
-    case resumeContinuationWithError(
-      upstreamContinuation: UnsafeContinuation<Void, Error>,
-      error: Error
+      upstreamContinuation: UnsafeContinuation<Void, Never>
     )
   }
 
   mutating func childTaskSuspended(
     baseIndex: Int,
-    continuation: UnsafeContinuation<Void, Error>
+    continuation: UnsafeContinuation<Void, Never>
   ) -> ChildTaskSuspendedAction? {
     switch self.state {
     case .initial:
@@ -193,9 +188,8 @@ struct CombineLatestManyStateMachine<Element: Sendable>: Sendable {
       // Since cancellation is cooperative it might be that child tasks are still getting
       // suspended even though we already cancelled them. We must tolerate this and just resume
       // the continuation with an error.
-      return .resumeContinuationWithError(
-        upstreamContinuation: continuation,
-        error: CancellationError()
+      return .resumeContinuation(
+        upstreamContinuation: continuation
       )
 
     case .modifying:
@@ -208,7 +202,7 @@ struct CombineLatestManyStateMachine<Element: Sendable>: Sendable {
     /// Indicates that the downstream continuation should be resumed with the element.
     case resumeContinuation(
       downstreamContinuation: DownstreamContinuation,
-      result: Result<[Element]?, Error>
+      result: Result<[Element]?, Failure>
     )
   }
 
@@ -293,14 +287,14 @@ struct CombineLatestManyStateMachine<Element: Sendable>: Sendable {
     /// Indicates the task and the upstream continuations should be cancelled.
     case cancelTaskAndUpstreamContinuations(
       task: Task<Void, Never>,
-      upstreamContinuations: [UnsafeContinuation<Void, Error>]
+      upstreamContinuations: [UnsafeContinuation<Void, Never>]
     )
     /// Indicates that the downstream continuation should be resumed with `nil` and
     /// the task and the upstream continuations should be cancelled.
     case resumeContinuationWithNilAndCancelTaskAndUpstreamContinuations(
       downstreamContinuation: DownstreamContinuation,
       task: Task<Void, Never>,
-      upstreamContinuations: [UnsafeContinuation<Void, Error>]
+      upstreamContinuations: [UnsafeContinuation<Void, Never>]
     )
   }
 
@@ -394,19 +388,19 @@ struct CombineLatestManyStateMachine<Element: Sendable>: Sendable {
     /// Indicates the task and the upstream continuations should be cancelled.
     case cancelTaskAndUpstreamContinuations(
       task: Task<Void, Never>,
-      upstreamContinuations: [UnsafeContinuation<Void, Error>]
+      upstreamContinuations: [UnsafeContinuation<Void, Never>]
     )
     /// Indicates that the downstream continuation should be resumed with the `error` and
     /// the task and the upstream continuations should be cancelled.
-    case resumeContinuationWithErrorAndCancelTaskAndUpstreamContinuations(
+    case resumeContinuationWithFailureAndCancelTaskAndUpstreamContinuations(
       downstreamContinuation: DownstreamContinuation,
-      error: Error,
+      error: Failure,
       task: Task<Void, Never>,
-      upstreamContinuations: [UnsafeContinuation<Void, Error>]
+      upstreamContinuations: [UnsafeContinuation<Void, Never>]
     )
   }
 
-  mutating func upstreamThrew(_ error: Error) -> UpstreamThrewAction? {
+  mutating func upstreamThrew(_ error: Failure) -> UpstreamThrewAction? {
     switch self.state {
     case .initial:
       preconditionFailure("Internal inconsistency current state \(self.state) and received upstreamThrew()")
@@ -433,7 +427,7 @@ struct CombineLatestManyStateMachine<Element: Sendable>: Sendable {
       // the upstream work.
       self.state = .finished
 
-      return .resumeContinuationWithErrorAndCancelTaskAndUpstreamContinuations(
+      return .resumeContinuationWithFailureAndCancelTaskAndUpstreamContinuations(
         downstreamContinuation: downstreamContinuation,
         error: error,
         task: task,
@@ -456,12 +450,12 @@ struct CombineLatestManyStateMachine<Element: Sendable>: Sendable {
     case resumeDownstreamContinuationWithNilAndCancelTaskAndUpstreamContinuations(
       downstreamContinuation: DownstreamContinuation,
       task: Task<Void, Never>,
-      upstreamContinuations: [UnsafeContinuation<Void, Error>]
+      upstreamContinuations: [UnsafeContinuation<Void, Never>]
     )
     /// Indicates that the task and the upstream continuations should be cancelled.
     case cancelTaskAndUpstreamContinuations(
       task: Task<Void, Never>,
-      upstreamContinuations: [UnsafeContinuation<Void, Error>]
+      upstreamContinuations: [UnsafeContinuation<Void, Never>]
     )
   }
 
@@ -515,12 +509,12 @@ struct CombineLatestManyStateMachine<Element: Sendable>: Sendable {
     case startTask([any Base])
     /// Indicates that all upstream continuations should be resumed.
     case resumeUpstreamContinuations(
-      upstreamContinuation: [UnsafeContinuation<Void, Error>]
+      upstreamContinuation: [UnsafeContinuation<Void, Never>]
     )
     /// Indicates that the downstream continuation should be resumed with the result.
     case resumeContinuation(
       downstreamContinuation: DownstreamContinuation,
-      result: Result<[Element]?, Error>
+      result: Result<[Element]?, Failure>
     )
     /// Indicates that the downstream continuation should be resumed with `nil`.
     case resumeDownstreamContinuationWithNil(DownstreamContinuation)
diff --git a/Sources/AsyncAlgorithms/CombineLatest/CombineLatestManyStorage.swift b/Sources/AsyncAlgorithms/CombineLatest/CombineLatestManyStorage.swift
@@ -10,8 +10,8 @@
 //===----------------------------------------------------------------------===//
 
 @available(AsyncAlgorithms 1.1, *)
-final class CombineLatestManyStorage<Element: Sendable>: Sendable {
-  typealias StateMachine = CombineLatestManyStateMachine<Element>
+final class CombineLatestManyStorage<Element: Sendable, Failure: Error>: Sendable {
+  typealias StateMachine = CombineLatestManyStateMachine<Element, Failure>
 
   private let stateMachine: ManagedCriticalState<StateMachine>
 
@@ -27,9 +27,8 @@ final class CombineLatestManyStorage<Element: Sendable>: Sendable {
       let task,
       let upstreamContinuation
     ):
-      upstreamContinuation.forEach { $0.resume(throwing: CancellationError()) }
-
       task.cancel()
+      upstreamContinuation.forEach { $0.resume() }
 
     case .none:
       break
@@ -97,14 +96,14 @@ final class CombineLatestManyStorage<Element: Sendable>: Sendable {
         let task,
         let upstreamContinuations
       ):
-        upstreamContinuations.forEach { $0.resume(throwing: CancellationError()) }
-        task.cancel()
+          task.cancel()
+        upstreamContinuations.forEach { $0.resume() }
 
         downstreamContinuation.resume(returning: .success(nil))
 
       case .cancelTaskAndUpstreamContinuations(let task, let upstreamContinuations):
-        upstreamContinuations.forEach { $0.resume(throwing: CancellationError()) }
-        task.cancel()
+          task.cancel()
+        upstreamContinuations.forEach { $0.resume() }
 
       case .none:
         break
@@ -114,13 +113,13 @@ final class CombineLatestManyStorage<Element: Sendable>: Sendable {
 
   private func startTask(
     stateMachine: inout StateMachine,
-    bases: [any CombineLatestManyStateMachine<Element>.Base],
+    bases: [any (AsyncSequence<Element, Failure> & Sendable)],
     downstreamContinuation: StateMachine.DownstreamContinuation
   ) {
     // This creates a new `Task` that is iterating the upstream
     // sequences. We must store it to cancel it at the right times.
     let task = Task {
-      await withThrowingTaskGroup(of: Void.self) { group in
+      await withTaskGroup(of: Result<Void, Failure>.self) { group in
         // For each upstream sequence we are adding a child task that
         // is consuming the upstream sequence
         for (baseIndex, base) in bases.enumerated() {
@@ -131,7 +130,7 @@ final class CombineLatestManyStorage<Element: Sendable>: Sendable {
               // We are creating a continuation before requesting the next
               // element from upstream. This continuation is only resumed
               // if the downstream consumer called `next` to signal his demand.
-              try await withUnsafeThrowingContinuation { continuation in
+              await withUnsafeContinuation { continuation in
                 let action = self.stateMachine.withCriticalRegion { stateMachine in
                   stateMachine.childTaskSuspended(baseIndex: baseIndex, continuation: continuation)
                 }
@@ -140,15 +139,19 @@ final class CombineLatestManyStorage<Element: Sendable>: Sendable {
                 case .resumeContinuation(let upstreamContinuation):
                   upstreamContinuation.resume()
 
-                case .resumeContinuationWithError(let upstreamContinuation, let error):
-                  upstreamContinuation.resume(throwing: error)
-
                 case .none:
                   break
                 }
               }
+              
+              let element: Element?
+              do {
+                element = try await baseIterator.next(isolation: nil)
+              } catch {
+                return .failure(error as! Failure) // Looks like a compiler bug
+              }
 
-              if let element = try await baseIterator.next() {
+              if let element = element {
                 let action = self.stateMachine.withCriticalRegion { stateMachine in
                   stateMachine.elementProduced(value: element, atBaseIndex: baseIndex)
                 }
@@ -172,15 +175,15 @@ final class CombineLatestManyStorage<Element: Sendable>: Sendable {
                   let upstreamContinuations
                 ):
 
-                  upstreamContinuations.forEach { $0.resume(throwing: CancellationError()) }
                   task.cancel()
+                  upstreamContinuations.forEach { $0.resume() }
 
                   downstreamContinuation.resume(returning: .success(nil))
                   break loop
 
                 case .cancelTaskAndUpstreamContinuations(let task, let upstreamContinuations):
-                  upstreamContinuations.forEach { $0.resume(throwing: CancellationError()) }
                   task.cancel()
+                  upstreamContinuations.forEach { $0.resume() }
 
                   break loop
 
@@ -189,30 +192,34 @@ final class CombineLatestManyStorage<Element: Sendable>: Sendable {
                 }
               }
             }
+            return .success(())
           }
         }
 
         while !group.isEmpty {
-          do {
-            try await group.next()
-          } catch {
+          let result = await group.next()
+          
+          switch result {
+          case .success, .none:
+            break
+          case .failure(let error):
             // One of the upstream sequences threw an error
             let action = self.stateMachine.withCriticalRegion { stateMachine in
               stateMachine.upstreamThrew(error)
             }
 
             switch action {
             case .cancelTaskAndUpstreamContinuations(let task, let upstreamContinuations):
-              upstreamContinuations.forEach { $0.resume(throwing: CancellationError()) }
               task.cancel()
-            case .resumeContinuationWithErrorAndCancelTaskAndUpstreamContinuations(
+              upstreamContinuations.forEach { $0.resume() }
+            case .resumeContinuationWithFailureAndCancelTaskAndUpstreamContinuations(
               let downstreamContinuation,
               let error,
               let task,
               let upstreamContinuations
             ):
-              upstreamContinuations.forEach { $0.resume(throwing: CancellationError()) }
               task.cancel()
+              upstreamContinuations.forEach { $0.resume() }
               downstreamContinuation.resume(returning: .failure(error))
             case .none:
               break
diff --git a/Tests/AsyncAlgorithmsTests/Support/GatedSequence.swift b/Tests/AsyncAlgorithmsTests/Support/GatedSequence.swift
@@ -10,6 +10,7 @@
 //===----------------------------------------------------------------------===//
 
 public struct GatedSequence<Element> {
+    typealias Failure = Never
   let elements: [Element]
   let gates: [Gate]
   var index = 0
diff --git a/Tests/AsyncAlgorithmsTests/TestCombineLatestMany.swift b/Tests/AsyncAlgorithmsTests/TestCombineLatestMany.swift
