fix(test): end the CI unit-test hang by making Turbine buffer emissions

.github/rulesets/protect-main.json

@@ -0,0 +1,26 @@
+{
+  "name": "Protect main",
+  "target": "branch",
+  "enforcement": "active",
+  "conditions": {
+    "ref_name": {
+      "include": ["~DEFAULT_BRANCH"],
+      "exclude": []
+    }
+  },
+  "bypass_actors": [],
+  "rules": [
+    {
+      "type": "required_status_checks",
+      "parameters": {
+        "strict_required_status_checks_policy": false,
+        "required_status_checks": [
+          { "context": "Unit tests (SwiftlyCore)" },
+          { "context": "Build app (iOS)" },
+          { "context": "Build app (macOS)" },
+          { "context": "Snapshot tests (iOS)" }
+        ]
+      }
+    }
+  ]
+}

SwiftlyCore/Sources/Common/Test/Turbine.swift

@@ -15,91 +15,98 @@ public func test<Value: Equatable & Sendable>(
       .removeDuplicates()
       .eraseToAnyPublisher()
   )
-  
+
   await block(turbine)
-  
+
   turbine.complete()
 }
 
 /// `Sendable` so a turbine can be awaited from the `@MainActor` test body without the compiler treating
 /// each `await` as sending it across an isolation boundary.
 public protocol Turbine<Value>: Sendable {
-  associatedtype Value
+  /// `Sendable` because values cross from the caller of these requirements into the main-actor-isolated
+  /// implementation; without the constraint the compiler rejects that hop for `any Turbine`. `test()`
+  /// already demands it of every value type.
+  associatedtype Value: Sendable
 
+  /// The oldest unconsumed emission, suspending until one arrives if none is buffered.
   func value() async -> Value
+
+  /// Consumes the first emission when it equals `value`; otherwise puts it back for the next `value()` call.
+  ///
+  /// A `@Published` publisher replays the current state on subscribe, so what arrives first depends on
+  /// timing: subscribe before the view model's startup work lands and it is the initial state; subscribe
+  /// after and the initial state was never seen, only the progressed one. Both orders are legitimate, which
+  /// is why a mismatch is put back rather than reported as a failure.
   func expectInitial(value: Value) async
 }
 
-/// `@unchecked` because the compiler cannot see that the only mutable state is a `CurrentValueSubject`
-/// (safe to send and read concurrently) and a `cancellables` array written once, during `init`.
-final class RealTurbine<Value: Equatable & Sendable>: Turbine, @unchecked Sendable {
+/// Buffers every emission into an `AsyncStream` and hands them out oldest-first, so waiting for the next
+/// value is a genuine suspension.
+///
+/// The previous implementation kept only the latest emission and, when `expectInitial` saw the expected
+/// value, recursed on unchanged state until something new arrived. That recursion has no suspension point,
+/// so it never yields its cooperative-pool thread and allocates async frames without bound. On a many-core
+/// dev machine the view model's next emission lands in milliseconds and the spin goes unnoticed; on a small
+/// CI runner a handful of tests spinning together occupy the whole pool, the work that would produce those
+/// emissions can never run, and the process starves until it dies — taking every suite's results with it.
+///
+/// `@MainActor` rather than `@unchecked Sendable`: every touchpoint already lives there — the publisher is
+/// subscribed and emits on the main actor, and the test body consuming the turbine runs on it too. Single
+/// consumer by contract, matching the sequential test bodies: two concurrent `value()` calls would race on
+/// the iterator.
+@MainActor
+final class RealTurbine<Value: Equatable & Sendable>: Turbine {
 
-  private let subject = CurrentValueSubject<TurbineValue<Value>, Never>(.notReady)
+  private var iterator: AsyncStream<Value>.Iterator
+  private let continuation: AsyncStream<Value>.Continuation
+  private var pushedBack: Value?
   private var cancellables: [AnyCancellable] = []
 
   init(publisher: AnyPublisher<Value, Never>) {
-    publisher.removeDuplicates()
-      .sink { value in self.subject.value = .ready(value) }
+    let (stream, continuation) = AsyncStream.makeStream(of: Value.self)
+    iterator = stream.makeAsyncIterator()
+    self.continuation = continuation
+    publisher
+      .removeDuplicates()
+      .sink { continuation.yield($0) }
       .store(in: &cancellables)
   }
 
   func value() async -> Value {
-    let value = switch subject.value {
-    case .notReady: await awaitFirst()
-    case let .ready(ready): ready
-    }
-    subject.value = .notReady
-    return value
+    await next()
   }
 
   func expectInitial(value: Value) async {
-    switch subject.value {
-    case .notReady:
-      let v = await awaitFirst()
-      if v == value {
-        await expectInitial(value: v)
-      }
-      subject.value = .notReady
-    case let .ready(v):
-      if v == value {
-        await expectInitial(value: v)
-      }
+    let first = await next()
+    if first != value {
+      pushedBack = first
     }
   }
 
   func complete() {
-    subject.send(completion: .finished)
+    continuation.finish()
     for cancellable in cancellables {
       cancellable.cancel()
     }
   }
 
-  private func awaitFirst() async -> Value {
-    await withUnsafeContinuation { continuation in
-      var cancellable: AnyCancellable?
-
-      // Unwrap to `.ready` values *before* `first()`. `subject` replays its current value on subscribe, and
-      // `awaitFirst` is only reached when that value is `.notReady` — so a bare `first()` delivers `.notReady`,
-      // takes the `break`, and completes without ever resuming the continuation. The awaiting task then stays
-      // suspended forever: the assertions still pass, but the process can never exit.
-      cancellable = subject
-        .compactMap { value -> Value? in
-          switch value {
-          case .notReady: nil
-          case let .ready(value): value
-          }
-        }
-        .first()
-        .sink { _ in
-          cancellable?.cancel()
-        } receiveValue: { value in
-          continuation.resume(returning: value)
-        }
+  private func next() async -> Value {
+    if let pushedBack {
+      self.pushedBack = nil
+      return pushedBack
+    }
+    // A `mutating async` call on an isolated stored property is rejected — the exclusive access would
+    // span the suspension — so advance a local copy; the iterator struct is only a handle to the
+    // stream's shared storage, and the write-back keeps the stored handle current. The advance goes
+    // through `next(isolation:)` because the plain `next()` is nonisolated and calling it would send
+    // this non-Sendable, main-actor-region iterator across an isolation boundary; passing the current
+    // isolation keeps the whole step on the main actor.
+    var iterator = self.iterator
+    guard let value = await iterator.next(isolation: #isolation) else {
+      fatalError("Turbine completed while a test was still awaiting a value")
     }
+    self.iterator = iterator
+    return value
   }
 }
-
-enum TurbineValue<V> {
-  case notReady
-  case ready(_ value: V)
-}

SwiftlyCore/Tests/Common/TestTests/TurbineTests.swift

@@ -0,0 +1,80 @@
+import Combine
+import SwiftlyTest
+import Testing
+
+@MainActor
+struct TurbineTests {
+
+  /// The minimal reproduction of the CI hang: `expectInitial` matching the current value used to spin
+  /// instead of consuming it, so this test never reached `send(1)`.
+  @Test
+  func whenSubscribedBeforeAnyChange_initialIsConsumed() async {
+    // given
+    let subject = CurrentValueSubject<Int, Never>(0)
+
+    await test(subject) { turbine in
+      await turbine.expectInitial(value: 0)
+
+      // when
+      subject.send(1)
+
+      // then
+      let result = await turbine.value()
+      #expect(result == 1)
+    }
+  }
+
+  @Test
+  func whenInitialWasMissed_firstValueIsPutBack() async {
+    // given: the observed source already progressed past its initial value before we subscribed
+    let subject = CurrentValueSubject<Int, Never>(5)
+
+    await test(subject) { turbine in
+      // when
+      await turbine.expectInitial(value: 0)
+
+      // then
+      let result = await turbine.value()
+      #expect(result == 5)
+    }
+  }
+
+  @Test
+  func valuesEmittedWhileNotAwaiting_areBufferedInOrder() async {
+    // given
+    let subject = CurrentValueSubject<Int, Never>(0)
+
+    await test(subject) { turbine in
+      await turbine.expectInitial(value: 0)
+
+      // when
+      subject.send(1)
+      subject.send(2)
+
+      // then
+      let first = await turbine.value()
+      let second = await turbine.value()
+      #expect(first == 1)
+      #expect(second == 2)
+    }
+  }
+
+  @Test
+  func whenValueArrivesWhileAwaiting_awaitResumesWithIt() async {
+    // given
+    let subject = CurrentValueSubject<Int, Never>(0)
+
+    await test(subject) { turbine in
+      await turbine.expectInitial(value: 0)
+
+      // when: emitted only after value() below has suspended and freed the main actor
+      Task {
+        subject.send(7)
+      }
+
+      // then
+      let result = await turbine.value()
+      #expect(result == 7)
+    }
+  }
+}