Audit pass on Sendable conformances and requirements

Author: phausler

Package.swift

@@ -16,11 +16,16 @@ let package = Package(
     .library(name: "_CAsyncSequenceValidationSupport", type: .static, targets: ["AsyncSequenceValidation"]),
     .library(name: "AsyncAlgorithms_XCTest", targets: ["AsyncAlgorithms_XCTest"]),
   ],
-  dependencies: [.package(url: "https://github.com/apple/swift-collections.git", .upToNextMajor(from: "1.0.3"))],
+  dependencies: [.package(url: "https://github.com/apple/swift-collections.git", .upToNextMajor(from: "1.0.4"))],
   targets: [
     .target(
       name: "AsyncAlgorithms",
-      dependencies: [.product(name: "Collections", package: "swift-collections")]
+      dependencies: [.product(name: "Collections", package: "swift-collections"),],
+      swiftSettings: [
+        .unsafeFlags([
+          "-strict-concurrency=complete"
+        ], .when(configuration: .debug))
+      ]
     ),
     .target(
       name: "AsyncSequenceValidation",

Sources/AsyncAlgorithms/Buffer/BoundedBufferStateMachine.swift

@@ -16,7 +16,7 @@ struct BoundedBufferStateMachine<Base: AsyncSequence> {
   typealias SuspendedProducer = UnsafeContinuation<Void, Never>
   typealias SuspendedConsumer = UnsafeContinuation<Result<Base.Element, Error>?, Never>
 
-  private enum State {
+  fileprivate enum State {
     case initial(base: Base)
     case buffering(
       task: Task<Void, Never>,
@@ -308,3 +308,6 @@ struct BoundedBufferStateMachine<Base: AsyncSequence> {
     }
   }
 }
+
+extension BoundedBufferStateMachine: Sendable where Base: Sendable { }
+extension BoundedBufferStateMachine.State: Sendable where Base: Sendable { }

Sources/AsyncAlgorithms/Buffer/UnboundedBufferStateMachine.swift

@@ -21,7 +21,7 @@ struct UnboundedBufferStateMachine<Base: AsyncSequence> {
     case bufferingOldest(Int)
   }
 
-  private enum State {
+  fileprivate enum State {
     case initial(base: Base)
     case buffering(
       task: Task<Void, Never>,
@@ -248,3 +248,6 @@ struct UnboundedBufferStateMachine<Base: AsyncSequence> {
     }
   }
 }
+
+extension UnboundedBufferStateMachine: Sendable where Base: Sendable { }
+extension UnboundedBufferStateMachine.State: Sendable where Base: Sendable { }

Sources/AsyncAlgorithms/Channels/AsyncChannel.swift

@@ -19,7 +19,7 @@
 /// on the `Iterator` is made, or when `finish()` is called from another Task.
 /// As `finish()` induces a terminal state, there is no more need for a back pressure management.
 /// This function does not suspend and will finish all the pending iterations.
-public final class AsyncChannel<Element>: AsyncSequence, @unchecked Sendable {
+public final class AsyncChannel<Element: Sendable>: AsyncSequence, @unchecked Sendable {
   public typealias Element = Element
   public typealias AsyncIterator = Iterator
 

Sources/AsyncAlgorithms/Channels/AsyncThrowingChannel.swift

@@ -18,7 +18,7 @@
 /// and is resumed when the next call to `next()` on the `Iterator` is made, or when `finish()`/`fail(_:)` is called
 /// from another Task. As `finish()` and `fail(_:)` induce a terminal state, there is no more need for a back pressure management.
 /// Those functions do not suspend and will finish all the pending iterations.
-public final class AsyncThrowingChannel<Element, Failure: Error>: AsyncSequence, @unchecked Sendable {
+public final class AsyncThrowingChannel<Element: Sendable, Failure: Error>: AsyncSequence, @unchecked Sendable {
   public typealias Element = Element
   public typealias AsyncIterator = Iterator
 

Sources/AsyncAlgorithms/Channels/ChannelStateMachine.swift

@@ -10,8 +10,42 @@
 //===----------------------------------------------------------------------===//
 @_implementationOnly import OrderedCollections
 
-struct ChannelStateMachine<Element, Failure: Error>: Sendable {
-  private struct SuspendedProducer: Hashable {
+struct OrderedSetContainer<Element: Hashable> {
+  var contents: OrderedSet<Element>
+  
+  var isEmpty: Bool { contents.isEmpty }
+  
+  mutating func removeFirst() -> Element  {
+    contents.removeFirst()
+  }
+  
+  mutating func remove(_ element: Element) -> Element? {
+    contents.remove(element)
+  }
+  
+  @discardableResult
+  mutating func append(_ element: Element) -> (inserted: Bool, index: Int) {
+    contents.append(element)
+  }
+  
+  func map<T>(_ apply: (Element) throws -> T) rethrows -> [T] {
+    try contents.map(apply)
+  }
+}
+
+extension OrderedSetContainer: ExpressibleByArrayLiteral {
+  typealias ArrayLiteralElement = OrderedSet<Element>.ArrayLiteralElement
+  
+  init(arrayLiteral elements: ArrayLiteralElement...) {
+    contents = OrderedSet(elements)
+  }
+}
+
+extension OrderedSetContainer: @unchecked Sendable where Element: Sendable { }
+
+
+struct ChannelStateMachine<Element: Sendable, Failure: Error>: Sendable {
+  private struct SuspendedProducer: Hashable, Sendable {
     let id: UInt64
     let continuation: UnsafeContinuation<Void, Never>?
     let element: Element?
@@ -29,7 +63,7 @@ struct ChannelStateMachine<Element, Failure: Error>: Sendable {
     }
   }
 
-  private struct SuspendedConsumer: Hashable {
+  private struct SuspendedConsumer: Hashable, Sendable {
     let id: UInt64
     let continuation: UnsafeContinuation<Element?, any Error>?
 
@@ -51,11 +85,11 @@ struct ChannelStateMachine<Element, Failure: Error>: Sendable {
     case failed(Error)
   }
 
-  private enum State {
+  private enum State: Sendable {
     case channeling(
-      suspendedProducers: OrderedSet<SuspendedProducer>,
+      suspendedProducers: OrderedSetContainer<SuspendedProducer>,
       cancelledProducers: Set<SuspendedProducer>,
-      suspendedConsumers: OrderedSet<SuspendedConsumer>,
+      suspendedConsumers: OrderedSetContainer<SuspendedConsumer>,
       cancelledConsumers: Set<SuspendedConsumer>
     )
     case terminated(Termination)

Sources/AsyncAlgorithms/Channels/ChannelStorage.swift

@@ -8,7 +8,7 @@
 // See https://swift.org/LICENSE.txt for license information
 //
 //===----------------------------------------------------------------------===//
-struct ChannelStorage<Element, Failure: Error>: Sendable {
+struct ChannelStorage<Element: Sendable, Failure: Error>: Sendable {
   private let stateMachine: ManagedCriticalState<ChannelStateMachine<Element, Failure>>
   private let ids = ManagedCriticalState<UInt64>(0)