Tolerate cancelling before we create our task (#250)

# Motivation
We switched the point in time when we create tasks to the first call to `next()`. This had the side effect that we can now be cancelled before we even created our task. Which by itself is fine but we didn't handle it in the state machines.

# Modification
Adapt the state machines of `debounce`, `zip`, `merge` and `combineLatest` to handle cancellation before `next()`.

# Result
We shouldn't run into these preconditions anymore.

Author: FranzBusch

Sources/AsyncAlgorithms/CombineLatest/CombineLatestStateMachine.swift

@@ -585,7 +585,9 @@ struct CombineLatestStateMachine<
   mutating func cancelled() -> CancelledAction? {
     switch self.state {
     case .initial:
-      preconditionFailure("Internal inconsistency current state \(self.state) and received cancelled()")
+      state = .finished
+
+      return .none
 
     case .waitingForDemand(let task, let upstreams, _):
       // The downstream task got cancelled so we need to cancel our upstream Task

Sources/AsyncAlgorithms/Debounce/DebounceStateMachine.swift

@@ -583,9 +583,8 @@ struct DebounceStateMachine<Base: AsyncSequence, C: Clock> {
     mutating func cancelled() -> CancelledAction? {
         switch self.state {
         case .initial:
-            // Since we are transitioning to `merging` before we return from `makeAsyncIterator`
-            // this can never happen
-            preconditionFailure("Internal inconsistency current state \(self.state) and received cancelled()")
+            state = .finished
+            return .none
 
         case .waitingForDemand:
             // We got cancelled before we event got any demand. This can happen if a cancelled task

Sources/AsyncAlgorithms/Merge/MergeStateMachine.swift

@@ -465,9 +465,11 @@ struct MergeStateMachine<
     mutating func cancelled() -> CancelledAction {
         switch state {
         case .initial:
-            // Since we are transitioning to `merging` before we return from `makeAsyncIterator`
-            // this can never happen
-            preconditionFailure("Internal inconsistency current state \(self.state) and received cancelled()")
+            // Since we are only transitioning to merging when the task is started we
+            // can be cancelled already.
+            state = .finished
+
+            return .none
 
         case let .merging(task, _, upstreamContinuations, _, .some(downstreamContinuation)):
             // The downstream Task got cancelled so we need to cancel our upstream Task

Sources/AsyncAlgorithms/Zip/ZipStateMachine.swift

@@ -460,7 +460,9 @@ struct ZipStateMachine<
   mutating func cancelled() -> CancelledAction? {
     switch self.state {
     case .initial:
-      preconditionFailure("Internal inconsistency current state \(self.state) and received cancelled()")
+        state = .finished
+
+        return .none
 
     case .waitingForDemand(let task, let upstreams):
       // The downstream task got cancelled so we need to cancel our upstream Task

Tests/AsyncAlgorithmsTests/TestCombineLatest.swift

@@ -318,6 +318,16 @@ final class TestCombineLatest2: XCTestCase {
     task.cancel()
     wait(for: [finished], timeout: 1.0)
   }
+
+    func test_combineLatest_when_cancelled() async {
+        let t = Task {
+          try? await Task.sleep(nanoseconds: 1_000_000_000)
+            let c1 = Indefinite(value: "test1").async
+            let c2 = Indefinite(value: "test1").async
+          for await _ in combineLatest(c1, c2) {}
+        }
+        t.cancel()
+    }
 }
 
 final class TestCombineLatest3: XCTestCase {

Tests/AsyncAlgorithmsTests/TestDebounce.swift

@@ -68,4 +68,15 @@ final class TestDebounce: XCTestCase {
         let throwingDebounce = [1].async.map { try throwOn(2, $0) }.debounce(for: .zero, clock: ContinuousClock())
         for try await _ in throwingDebounce {}
     }
+
+  func test_debounce_when_cancelled() async throws {
+      guard #available(macOS 13.0, iOS 16.0, watchOS 9.0, tvOS 16.0, *) else { throw XCTSkip("Skipped due to Clock/Instant/Duration availability") }
+
+      let t = Task {
+        try? await Task.sleep(nanoseconds: 1_000_000_000)
+          let c1 = Indefinite(value: "test1").async
+          for await _ in c1.debounce(for: .seconds(1), clock: .continuous) {}
+      }
+      t.cancel()
+  }
 }

Tests/AsyncAlgorithmsTests/TestMerge.swift

@@ -191,6 +191,16 @@ final class TestMerge2: XCTestCase {
     task.cancel()
     wait(for: [finished], timeout: 1.0)
   }
+
+    func test_merge_when_cancelled() async {
+        let t = Task {
+          try? await Task.sleep(nanoseconds: 1_000_000_000)
+            let c1 = Indefinite(value: "test1").async
+            let c2 = Indefinite(value: "test1").async
+          for await _ in merge(c1, c2) {}
+        }
+        t.cancel()
+    }
 }
 
 final class TestMerge3: XCTestCase {

Tests/AsyncAlgorithmsTests/TestZip.swift

@@ -158,6 +158,16 @@ final class TestZip2: XCTestCase {
     task.cancel()
     wait(for: [finished], timeout: 1.0)
   }
+
+  func test_zip_when_cancelled() async {
+      let t = Task {
+        try? await Task.sleep(nanoseconds: 1_000_000_000)
+          let c1 = Indefinite(value: "test1").async
+          let c2 = Indefinite(value: "test1").async
+        for await _ in zip(c1, c2) {}
+      }
+      t.cancel()
+  }
 }
 
 final class TestZip3: XCTestCase {