DuplexAsyncChannel/MPSC cancellation hang

[ehaydenr]

I'm using the DuplexAsyncChannel and finding that sometimes the code just hangs. The following is what I believe to be a reproduction of that hang.

#if UnstableAsyncStreaming && compiler(>=6.4)

import AsyncStreaming

#if canImport(Glibc)
import Glibc
#elseif canImport(Darwin)
import Darwin
#endif

@available(macOS 15.0, iOS 18.0, tvOS 18.0, visionOS 2.0, *)
@main
struct MinimalRepro {
  static func main() async {
    await withTaskGroup(of: Void.self) { group in
      // A child added to an already-cancelled group is cancelled at creation, so
      // the read inside `forEachBuffer` starts out cancelled — no polling, no
      // sleeping, and no dependence on which task the executor runs first.
      group.cancelAll()
      group.addTask {
        await DuplexAsyncChannel<Int, Void, Never>.withDuplex(
          of: Int.self,
          backpressureStrategy: .watermark(low: 1, high: 2)
        ) { writerA, readerA, writerB, readerB in
          _ = consume readerA
          _ = consume writerB
          // Dropped unfinished, so the forward direction stays open and empty:
          // the read cannot complete and has to park.
          _ = consume writerA

          print("forEachBuffer on an already-cancelled reader must throw CancellationError.")
          print("If this process is still alive, it did not: it is spinning. Ctrl-C to stop.")
          // Flushed because the spin below never returns, and a piped stdout would
          // otherwise never be drained.
          fflush(nil)

          do {
            // Loops until a read reports a non-`nil` final element. The lost
            // cancellation means that never happens, and because each read
            // completes without suspending, this pegs a core and stops observing
            // cancellation. `body` is never invoked: `forEachBuffer` skips it for
            // an empty buffer, and every chunk here is empty.
            _ = try await readerB.forEachBuffer { _ in
              print("body invoked")
            }
            print("forEachBuffer returned")
          } catch {
            print("OK: forEachBuffer threw \(error)")
          }
        }
      }
    }
  }
}

#endif