From 49b17bcf9d95c3592063f300780256dd28bf48be Mon Sep 17 00:00:00 2001 From: David Reinecke Date: Sun, 25 Oct 2026 17:29:48 +0000 Subject: [PATCH 1/4] DeveloperServices: assign all of an app's groups to its App ID in one request assignApplicationGroupToAppId replaces the App ID's groups with the ones in the request, and DeveloperServicesAssignAppGroupsOperation sent one request per group, concurrently. An app whose entitlements name two groups therefore ended with one group in its profile (whichever request Apple handled last), and the device refused to install it (ApplicationVerificationFailed). The operation now creates any missing groups as before, then sends a single request carrying every group ID, and returns the group IDs in the order the entitlements listed them, so the rewritten entitlement is stable between builds. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01UKzxgxaLwK7Ap5LMwj9MxH --- ...veloperServicesAssignAppGroupRequest.swift | 11 ++-- ...operServicesAssignAppGroupsOperation.swift | 60 +++++++++++-------- 2 files changed, 41 insertions(+), 30 deletions(-) diff --git a/Sources/XKit/DeveloperServices/App Groups/DeveloperServicesAssignAppGroupRequest.swift b/Sources/XKit/DeveloperServices/App Groups/DeveloperServicesAssignAppGroupRequest.swift index 713afdb0..399ed8c6 100644 --- a/Sources/XKit/DeveloperServices/App Groups/DeveloperServicesAssignAppGroupRequest.swift +++ b/Sources/XKit/DeveloperServices/App Groups/DeveloperServicesAssignAppGroupRequest.swift @@ -8,6 +8,9 @@ import Foundation +/// Assigns a set of app groups to an App ID. The request replaces the App ID's groups with +/// `groupIDs`, so every group the app needs has to be in one request: assigning them one +/// request at a time leaves only the last one standing. public final class DeveloperServicesAssignAppGroupRequest: DeveloperServicesPlatformRequest { public struct Response: Decodable, Sendable {} @@ -16,14 +19,14 @@ public final class DeveloperServicesAssignAppGroupRequest: DeveloperServicesPlat public let platform: DeveloperServicesPlatform public let teamID: DeveloperServicesTeam.ID public let appIDID: String - public let groupID: DeveloperServicesAppGroup.ID + public let groupIDs: [DeveloperServicesAppGroup.ID] var subAction: String { return "assignApplicationGroupToAppId" } var subParameters: [String: Any] { return [ "teamId": teamID.rawValue, "appIdId": appIDID, - "applicationGroups": groupID.rawValue + "applicationGroups": groupIDs.map(\.rawValue) ] } @@ -31,12 +34,12 @@ public final class DeveloperServicesAssignAppGroupRequest: DeveloperServicesPlat platform: DeveloperServicesPlatform, teamID: DeveloperServicesTeam.ID, appIDID: String, - groupID: DeveloperServicesAppGroup.ID + groupIDs: [DeveloperServicesAppGroup.ID] ) { self.platform = platform self.teamID = teamID self.appIDID = appIDID - self.groupID = groupID + self.groupIDs = groupIDs } } diff --git a/Sources/XKit/DeveloperServices/App Groups/DeveloperServicesAssignAppGroupsOperation.swift b/Sources/XKit/DeveloperServices/App Groups/DeveloperServicesAssignAppGroupsOperation.swift index b13e1a5b..d10fa846 100644 --- a/Sources/XKit/DeveloperServices/App Groups/DeveloperServicesAssignAppGroupsOperation.swift +++ b/Sources/XKit/DeveloperServices/App Groups/DeveloperServicesAssignAppGroupsOperation.swift @@ -33,52 +33,60 @@ public struct DeveloperServicesAssignAppGroupsOperation: DeveloperServicesOperat self.client = DeveloperServicesClient(authData: authData) } + /// Finds the team's group for `groupID`, creating it when there is none. private func upsertAppGroup( _ groupID: DeveloperServicesAppGroup.GroupID, existingGroups: [String: DeveloperServicesAppGroup] - ) async throws -> DeveloperServicesAppGroup.GroupID { + ) async throws -> DeveloperServicesAppGroup { let sanitized = ProvisioningIdentifiers.sanitize(groupID: groupID) - let group: DeveloperServicesAppGroup if let existingGroup = existingGroups[sanitized] { - group = existingGroup - } else { - let groupID = ProvisioningIdentifiers.groupID(fromSanitized: sanitized, context: context) - let name = ProvisioningIdentifiers.groupName(fromSanitized: sanitized) - let request = DeveloperServicesAddAppGroupRequest( - platform: .iOS, - teamID: xcodeAuthData.teamID, - name: name, - groupID: groupID - ) - group = try await client.send(request) + return existingGroup } - - _ = try await client.send( - DeveloperServicesAssignAppGroupRequest( - platform: .iOS, - teamID: xcodeAuthData.teamID, - appIDID: appID.id, - groupID: group.id - ) + let groupID = ProvisioningIdentifiers.groupID(fromSanitized: sanitized, context: context) + let name = ProvisioningIdentifiers.groupName(fromSanitized: sanitized) + let request = DeveloperServicesAddAppGroupRequest( + platform: .iOS, + teamID: xcodeAuthData.teamID, + name: name, + groupID: groupID ) - - return group.groupID + return try await client.send(request) } + /// Creates any missing groups, then assigns all of them to the App ID in one request. + /// `assignApplicationGroupToAppId` replaces the App ID's groups with the ones in the request, + /// so assigning each group separately (as this once did, concurrently) left the App ID with + /// only one of them, and an app with two groups got a profile that iOS refused to install. + /// Returns the groups' IDs in the order they were given. public func perform() async throws -> [DeveloperServicesAppGroup.GroupID] { let existing = try await client.send(DeveloperServicesListAppGroupsRequest( platform: .iOS, teamID: xcodeAuthData.teamID )) let sanitized = existing.map { (ProvisioningIdentifiers.sanitize(groupID: $0.groupID), $0) } let dict = Dictionary(sanitized, uniquingKeysWith: { $1 }) - return try await withThrowingTaskGroup(of: DeveloperServicesAppGroup.GroupID.self) { group in - for groupID in groupIDs { + let groups = try await withThrowingTaskGroup(of: (Int, DeveloperServicesAppGroup).self) { group in + for (index, groupID) in groupIDs.enumerated() { group.addTask { - try await upsertAppGroup(groupID, existingGroups: dict) + (index, try await upsertAppGroup(groupID, existingGroups: dict)) } } return try await group.reduce(into: []) { $0.append($1) } } + .sorted { $0.0 < $1.0 } + .map(\.1) + + if !groups.isEmpty { + _ = try await client.send( + DeveloperServicesAssignAppGroupRequest( + platform: .iOS, + teamID: xcodeAuthData.teamID, + appIDID: appID.id, + groupIDs: groups.map(\.id) + ) + ) + } + + return groups.map(\.groupID) } } From 2d3c5d59006e94faf148a5f72a5b4172e11d74a6 Mon Sep 17 00:00:00 2001 From: Kabir Oberai Date: Fri, 30 Oct 2026 00:17:36 -0400 Subject: [PATCH 2/4] Clean up --- ...veloperServicesAssignAppGroupRequest.swift | 4 +-- ...operServicesAssignAppGroupsOperation.swift | 26 +++++++------------ 2 files changed, 10 insertions(+), 20 deletions(-) diff --git a/Sources/XKit/DeveloperServices/App Groups/DeveloperServicesAssignAppGroupRequest.swift b/Sources/XKit/DeveloperServices/App Groups/DeveloperServicesAssignAppGroupRequest.swift index 399ed8c6..a0d4c9e7 100644 --- a/Sources/XKit/DeveloperServices/App Groups/DeveloperServicesAssignAppGroupRequest.swift +++ b/Sources/XKit/DeveloperServices/App Groups/DeveloperServicesAssignAppGroupRequest.swift @@ -8,9 +8,7 @@ import Foundation -/// Assigns a set of app groups to an App ID. The request replaces the App ID's groups with -/// `groupIDs`, so every group the app needs has to be in one request: assigning them one -/// request at a time leaves only the last one standing. +/// Replaces all of the app's groups with the given ones public final class DeveloperServicesAssignAppGroupRequest: DeveloperServicesPlatformRequest { public struct Response: Decodable, Sendable {} diff --git a/Sources/XKit/DeveloperServices/App Groups/DeveloperServicesAssignAppGroupsOperation.swift b/Sources/XKit/DeveloperServices/App Groups/DeveloperServicesAssignAppGroupsOperation.swift index d10fa846..85dead49 100644 --- a/Sources/XKit/DeveloperServices/App Groups/DeveloperServicesAssignAppGroupsOperation.swift +++ b/Sources/XKit/DeveloperServices/App Groups/DeveloperServicesAssignAppGroupsOperation.swift @@ -33,8 +33,7 @@ public struct DeveloperServicesAssignAppGroupsOperation: DeveloperServicesOperat self.client = DeveloperServicesClient(authData: authData) } - /// Finds the team's group for `groupID`, creating it when there is none. - private func upsertAppGroup( + private func getOrCreateAppGroup( _ groupID: DeveloperServicesAppGroup.GroupID, existingGroups: [String: DeveloperServicesAppGroup] ) async throws -> DeveloperServicesAppGroup { @@ -53,11 +52,6 @@ public struct DeveloperServicesAssignAppGroupsOperation: DeveloperServicesOperat return try await client.send(request) } - /// Creates any missing groups, then assigns all of them to the App ID in one request. - /// `assignApplicationGroupToAppId` replaces the App ID's groups with the ones in the request, - /// so assigning each group separately (as this once did, concurrently) left the App ID with - /// only one of them, and an app with two groups got a profile that iOS refused to install. - /// Returns the groups' IDs in the order they were given. public func perform() async throws -> [DeveloperServicesAppGroup.GroupID] { let existing = try await client.send(DeveloperServicesListAppGroupsRequest( platform: .iOS, teamID: xcodeAuthData.teamID @@ -67,7 +61,7 @@ public struct DeveloperServicesAssignAppGroupsOperation: DeveloperServicesOperat let groups = try await withThrowingTaskGroup(of: (Int, DeveloperServicesAppGroup).self) { group in for (index, groupID) in groupIDs.enumerated() { group.addTask { - (index, try await upsertAppGroup(groupID, existingGroups: dict)) + (index, try await getOrCreateAppGroup(groupID, existingGroups: dict)) } } return try await group.reduce(into: []) { $0.append($1) } @@ -75,16 +69,14 @@ public struct DeveloperServicesAssignAppGroupsOperation: DeveloperServicesOperat .sorted { $0.0 < $1.0 } .map(\.1) - if !groups.isEmpty { - _ = try await client.send( - DeveloperServicesAssignAppGroupRequest( - platform: .iOS, - teamID: xcodeAuthData.teamID, - appIDID: appID.id, - groupIDs: groups.map(\.id) - ) + _ = try await client.send( + DeveloperServicesAssignAppGroupRequest( + platform: .iOS, + teamID: xcodeAuthData.teamID, + appIDID: appID.id, + groupIDs: groups.map(\.id) ) - } + ) return groups.map(\.groupID) } From 676e3516cb1acd25a80b62011d54bec37d78675b Mon Sep 17 00:00:00 2001 From: Kabir Oberai Date: Fri, 30 Oct 2026 00:57:04 -0400 Subject: [PATCH 3/4] Add concurrentMap --- ...operServicesAssignAppGroupsOperation.swift | 12 +- Sources/XUtils/Stdlib+Utils.swift | 31 ++++ Tests/XToolTests/ConcurrentMapTests.swift | 161 ++++++++++++++++++ 3 files changed, 195 insertions(+), 9 deletions(-) create mode 100644 Sources/XUtils/Stdlib+Utils.swift create mode 100644 Tests/XToolTests/ConcurrentMapTests.swift diff --git a/Sources/XKit/DeveloperServices/App Groups/DeveloperServicesAssignAppGroupsOperation.swift b/Sources/XKit/DeveloperServices/App Groups/DeveloperServicesAssignAppGroupsOperation.swift index 85dead49..5516bd10 100644 --- a/Sources/XKit/DeveloperServices/App Groups/DeveloperServicesAssignAppGroupsOperation.swift +++ b/Sources/XKit/DeveloperServices/App Groups/DeveloperServicesAssignAppGroupsOperation.swift @@ -8,6 +8,7 @@ import Foundation import DeveloperAPI +import XUtils public struct DeveloperServicesAssignAppGroupsOperation: DeveloperServicesOperation { @@ -58,16 +59,9 @@ public struct DeveloperServicesAssignAppGroupsOperation: DeveloperServicesOperat )) let sanitized = existing.map { (ProvisioningIdentifiers.sanitize(groupID: $0.groupID), $0) } let dict = Dictionary(sanitized, uniquingKeysWith: { $1 }) - let groups = try await withThrowingTaskGroup(of: (Int, DeveloperServicesAppGroup).self) { group in - for (index, groupID) in groupIDs.enumerated() { - group.addTask { - (index, try await getOrCreateAppGroup(groupID, existingGroups: dict)) - } - } - return try await group.reduce(into: []) { $0.append($1) } + let groups = try await groupIDs.concurrentMap { groupID in + try await getOrCreateAppGroup(groupID, existingGroups: dict) } - .sorted { $0.0 < $1.0 } - .map(\.1) _ = try await client.send( DeveloperServicesAssignAppGroupRequest( diff --git a/Sources/XUtils/Stdlib+Utils.swift b/Sources/XUtils/Stdlib+Utils.swift new file mode 100644 index 00000000..487ebf6f --- /dev/null +++ b/Sources/XUtils/Stdlib+Utils.swift @@ -0,0 +1,31 @@ +extension Sequence where Element: Sendable { + public nonisolated(nonsending) func concurrentMap( + _ transform: @escaping @Sendable @isolated(any) (Element) async throws(E) -> T + ) async throws(E) -> [T] { + try await withTaskGroup(of: (Int, Result).self) { group -> Result<[T], E> in + var results: [T?] = [] + results.reserveCapacity(self.underestimatedCount) + for (index, element) in enumerated() { + group.addTask { + let result = await Result { () async throws(E) in + try await transform(element) + } + return (index, result) + } + results.append(nil) + } + for await (index, result) in group { + switch result { + case .success(let value): + results[index] = value + case .failure(let error): + group.cancelAll() + return .failure(error) + } + } + // we can't use `as! [T]` because it breaks nested optionals (Foo??) + return .success(results.map { $0! }) + } + .get() + } +} diff --git a/Tests/XToolTests/ConcurrentMapTests.swift b/Tests/XToolTests/ConcurrentMapTests.swift new file mode 100644 index 00000000..c4130930 --- /dev/null +++ b/Tests/XToolTests/ConcurrentMapTests.swift @@ -0,0 +1,161 @@ +import Testing +import XUtils + +@Suite(.timeLimit(.minutes(1))) +struct ConcurrentMapTests { + @Test(arguments: [[], [7], [3, 1, 3, 2], Array(0..<10)]) + func nonThrowingTransform(input: [Int]) async { + // No `try` is needed when the transform cannot throw. + let result = await input.concurrentMap { String($0 * 2) } + #expect(result == input.map { String($0 * 2) }) + } + + @Test func emptyInputDoesNotCallTransform() async throws { + let result = try await [Int]().concurrentMap { _ throws(MapError) -> Int in + Issue.record("An empty sequence must not invoke the transform") + throw .failed(0) + } + #expect(result.isEmpty) + } + + @Test func singlePassSequence() async { + var next = 0 + let input = AnySequence(AnyIterator { + defer { next += 1 } + return next < 4 ? next : nil + }) + #expect(input.underestimatedCount == 0) + + let result = await input.concurrentMap { $0 * 2 } + + #expect(result == [0, 2, 4, 6]) + } + + @Test func optionalResults() async { + let result = await (0..<4).concurrentMap { value -> Int? in + value.isMultiple(of: 2) ? nil : value + } + #expect(result == [nil, 1, nil, 3]) + } + + @Test func nestedOptionalResults() async { + let input: [Int??] = [.none, .some(.none), .some(.some(3))] + let result = await input.concurrentMap { $0 } + #expect(result == input) + } + + @Test func concurrentExecutionPreservesInputOrder() async { + let started = AsyncStream.makeStream() + let finished = AsyncStream.makeStream() + let gates = (0..<4).map { _ in AsyncStream.makeStream() } + let task = Task { + await (0..<4).concurrentMap { value in + started.continuation.yield(value) + for await _ in gates[value].stream {} + finished.continuation.yield(value) + return value * 2 + } + } + defer { task.cancel() } + + // Every transform must start before any is allowed to finish. + var starts = started.stream.makeAsyncIterator() + var startedValues: [Int] = [] + for _ in 0..<4 { + if let value = await starts.next() { + startedValues.append(value) + } + } + #expect(startedValues.sorted() == [0, 1, 2, 3]) + + // Let transforms finish in reverse order, independently of scheduler timing. + var finishes = finished.stream.makeAsyncIterator() + for value in (0..<4).reversed() { + gates[value].continuation.finish() + #expect(await finishes.next() == value) + } + #expect(await task.value == [0, 2, 4, 6]) + } + + @Test func preservesTypedError() async { + // This signature also checks that concurrentMap preserves the error type. + func map() async throws(MapError) -> [Int] { + try await [42].concurrentMap { value throws(MapError) -> Int in + throw .failed(value) + } + } + + await #expect(throws: MapError.failed(42)) { + try await map() + } + } + + @Test func failureCancelsUnfinishedTransforms() async { + let started = AsyncStream.makeStream() + let failureGate = AsyncStream.makeStream() + let siblingGate = AsyncStream.makeStream() + let cancelled = AsyncStream.makeStream() + let task = Task { + try await (0..<3).concurrentMap { value async throws(MapError) -> Int in + started.continuation.yield(value) + if value == 1 { + for await _ in failureGate.stream {} + throw .failed(value) + } + // This stream only finishes when the task is cancelled. + for await _ in siblingGate.stream {} + #expect(Task.isCancelled) + cancelled.continuation.yield(value) + return value + } + } + defer { task.cancel() } + + var starts = started.stream.makeAsyncIterator() + for _ in 0..<3 { + _ = await starts.next() + } + failureGate.continuation.finish() + + await #expect(throws: MapError.failed(1)) { + try await withTaskCancellationHandler { + try await task.value + } onCancel: { + // Let the suite's time limit break a regression in sibling cancellation. + task.cancel() + } + } + // The task group must wait for both cancelled transforms to exit. + cancelled.continuation.finish() + var cancelledValues: [Int] = [] + for await value in cancelled.stream { + cancelledValues.append(value) + } + #expect(cancelledValues.sorted() == [0, 2]) + } + + @Test func parentCancellationReachesTransforms() async { + let started = AsyncStream.makeStream() + let gate = AsyncStream.makeStream() + let task = Task { + try await [1].concurrentMap { _ in + started.continuation.yield(()) + for await _ in gate.stream {} + try Task.checkCancellation() + } + } + defer { task.cancel() } + + var starts = started.stream.makeAsyncIterator() + _ = await starts.next() + task.cancel() + + await #expect(throws: CancellationError.self) { + try await task.value + } + } + + private enum MapError: Error, Equatable { + case failed(Int) + } +} From c9032326f8be77d0d8226a7c40e3c334e746381c Mon Sep 17 00:00:00 2001 From: Kabir Oberai Date: Fri, 30 Oct 2026 01:29:04 -0400 Subject: [PATCH 4/4] Concurrency limit --- Sources/XUtils/Stdlib+Utils.swift | 55 +++++++++++--- Tests/XToolTests/ConcurrentMapTests.swift | 92 +++++++++++++++++++---- 2 files changed, 124 insertions(+), 23 deletions(-) diff --git a/Sources/XUtils/Stdlib+Utils.swift b/Sources/XUtils/Stdlib+Utils.swift index 487ebf6f..5084948d 100644 --- a/Sources/XUtils/Stdlib+Utils.swift +++ b/Sources/XUtils/Stdlib+Utils.swift @@ -1,20 +1,55 @@ extension Sequence where Element: Sendable { + /// Like `map`, but runs transformations concurrently. + /// + /// - Parameter maxConcurrency: The maximum number of transforms to run concurrently. + /// If `nil`, we choose a reasonable default value. Use `.max` to run all at once. + /// + /// - Parameter transform: The transformation to apply to each element. + /// + /// - Returns: Transformed elements in the same order as the original sequence. public nonisolated(nonsending) func concurrentMap( - _ transform: @escaping @Sendable @isolated(any) (Element) async throws(E) -> T + maxConcurrency: Int? = nil, + _ transform: @escaping @Sendable @isolated(any) (Element) async throws(E) -> T, ) async throws(E) -> [T] { - try await withTaskGroup(of: (Int, Result).self) { group -> Result<[T], E> in + try await _concurrentMap( + maxConcurrency: maxConcurrency ?? 10, + transform, + willWaitForNext: nil, + ) + } + + nonisolated(nonsending) func _concurrentMap( + maxConcurrency: Int, + _ transform: @escaping @Sendable @isolated(any) (Element) async throws(E) -> T, + willWaitForNext: (() -> Void)?, // hook for testing + ) async throws(E) -> [T] { + precondition(maxConcurrency > 0, "maxConcurrency must be > 0") + + return try await withTaskGroup(of: (Int, Result).self) { group -> Result<[T], E> in var results: [T?] = [] results.reserveCapacity(self.underestimatedCount) - for (index, element) in enumerated() { - group.addTask { - let result = await Result { () async throws(E) in - try await transform(element) + var iterator = makeIterator() + var exhausted = false + var inFlight = 0 + while true { + while !exhausted && inFlight < maxConcurrency { + guard let element = iterator.next() else { + exhausted = true + break } - return (index, result) + let index = results.count + results.append(nil) + group.addTask { + let result = await Result { () async throws(E) in + try await transform(element) + } + return (index, result) + } + inFlight += 1 } - results.append(nil) - } - for await (index, result) in group { + willWaitForNext?() + guard let (index, result) = await group.next() else { break } + inFlight -= 1 switch result { case .success(let value): results[index] = value diff --git a/Tests/XToolTests/ConcurrentMapTests.swift b/Tests/XToolTests/ConcurrentMapTests.swift index c4130930..32456b7c 100644 --- a/Tests/XToolTests/ConcurrentMapTests.swift +++ b/Tests/XToolTests/ConcurrentMapTests.swift @@ -1,12 +1,12 @@ import Testing -import XUtils +@testable import XUtils @Suite(.timeLimit(.minutes(1))) struct ConcurrentMapTests { - @Test(arguments: [[], [7], [3, 1, 3, 2], Array(0..<10)]) - func nonThrowingTransform(input: [Int]) async { + @Test(arguments: [[], [7], [3, 1, 3, 2], Array(0..<10)], [nil, 1, 2, Int.max] as [Int?]) + func nonThrowingTransform(input: [Int], maxConcurrency: Int?) async { // No `try` is needed when the transform cannot throw. - let result = await input.concurrentMap { String($0 * 2) } + let result = await input.concurrentMap(maxConcurrency: maxConcurrency) { String($0 * 2) } #expect(result == input.map { String($0 * 2) }) } @@ -26,7 +26,7 @@ struct ConcurrentMapTests { }) #expect(input.underestimatedCount == 0) - let result = await input.concurrentMap { $0 * 2 } + let result = await input.concurrentMap(maxConcurrency: 1) { $0 * 2 } #expect(result == [0, 2, 4, 6]) } @@ -44,12 +44,13 @@ struct ConcurrentMapTests { #expect(result == input) } - @Test func concurrentExecutionPreservesInputOrder() async { + @Test(arguments: [2, 4, Int.max]) + func concurrentExecutionPreservesInputOrder(maxConcurrency: Int) async { let started = AsyncStream.makeStream() let finished = AsyncStream.makeStream() let gates = (0..<4).map { _ in AsyncStream.makeStream() } let task = Task { - await (0..<4).concurrentMap { value in + await (0..<4).concurrentMap(maxConcurrency: maxConcurrency) { value in started.continuation.yield(value) for await _ in gates[value].stream {} finished.continuation.yield(value) @@ -58,25 +59,89 @@ struct ConcurrentMapTests { } defer { task.cancel() } - // Every transform must start before any is allowed to finish. + // Fill the concurrency window before allowing any transform to finish. + let initialCount = min(maxConcurrency, 4) var starts = started.stream.makeAsyncIterator() var startedValues: [Int] = [] - for _ in 0..<4 { + for _ in 0...makeStream() + let gates = (0..<25).map { _ in AsyncStream.makeStream() } + let task = Task { + // Both closures below run in the producer task. Only immutable snapshots + // cross to the test through the checkpoint stream. + var submitted = 0 + let input = AnySequence(AnyIterator { + guard submitted < gates.count else { return nil } + defer { submitted += 1 } + return submitted + }) + return await input._concurrentMap(maxConcurrency: maxConcurrency) { value in + for await _ in gates[value].stream {} + return value * 2 + } willWaitForNext: { + checkpoints.continuation.yield(submitted) + } + } + defer { task.cancel() } + + // No transform can finish yet. The hook proves submission has reached + // group.next(), rather than merely being paused by the scheduler. + let initialCount = min(maxConcurrency, gates.count) + var waits = checkpoints.stream.makeAsyncIterator() + #expect(await waits.next() == initialCount) + + // Each completed transform frees exactly one slot. Check the next submission + // boundary before releasing another transform, including while draining. + for value in gates.indices { + gates[value].continuation.finish() + #expect(await waits.next() == min(initialCount + value + 1, gates.count)) + } + let result = await withTaskCancellationHandler { + await task.value + } onCancel: { + task.cancel() + } + + #expect(result == (0..<25).map { $0 * 2 }) + } + + @Test @MainActor func actorIsolatedTransform() async { + var visited: [Int] = [] + let result = await (0..<4).concurrentMap(maxConcurrency: 2) { @MainActor value in + MainActor.assertIsolated() + visited.append(value) + return value * 2 + } + + #expect(visited.sorted() == [0, 1, 2, 3]) + #expect(result == [0, 2, 4, 6]) + } + @Test func preservesTypedError() async { // This signature also checks that concurrentMap preserves the error type. func map() async throws(MapError) -> [Int] { @@ -96,7 +161,8 @@ struct ConcurrentMapTests { let siblingGate = AsyncStream.makeStream() let cancelled = AsyncStream.makeStream() let task = Task { - try await (0..<3).concurrentMap { value async throws(MapError) -> Int in + try await (0..<10).concurrentMap(maxConcurrency: 3) { value async throws(MapError) -> Int in + #expect(value < 3, "A failure must not schedule the remaining inputs") started.continuation.yield(value) if value == 1 { for await _ in failureGate.stream {}