NonisolatedNonsendingByDefault and other concurrency fixes (#63)

### Motivation

- Fixes #17
- Fixes #23
- Fixes #32
- Fixes #43
- Fixes #44

These 5 fixes are intertwined and were difficult to split out, so here's
one PR that fixes them all.

### Modifications

- The PR enables NonisolatedNonsendingByDefault, which caused an error
in the existing combineLatest implementation, which we wanted to replace
anyway.
- So we vendor in the code from Swift Async Algoritms (the PR is open
here: apple/swift-async-algorithms#360), the
plan is to remove this copy when it lands in async algos.
- Updating the combineLatest implementation to the correct one requires
the async sequences to be Sendable, which in turn requires Swift 6.2 for
us to be able to spell `any (AsyncSequence & Sendable)` correctly, it
doesn't work on 6.1.
- We also enabled the explicit sendable warning in CI, helping us ensure
all our public API is explicitly annotated as Sendable or not.

### Result

Addressed 5 important issues that impact the API and concurrency.

### Test Plan

Brought over tests for the new combineLatest implementation, added more
unit tests for failure cases when using the ConfigReader with multiple
providers.

Author: czechboy0

Package.swift

@@ -61,6 +61,7 @@ let enableAllTraitsExplicit = ProcessInfo.processInfo.environment["ENABLE_ALL_TR
 
 let enableAllTraits = spiGenerateDocs || previewDocs || enableAllTraitsExplicit
 let addDoccPlugin = previewDocs || spiGenerateDocs
+let enableAllCIFlags = enableAllTraitsExplicit
 
 traits.insert(
     .default(
@@ -77,6 +78,7 @@ let package = Package(
     traits: traits,
     dependencies: [
         .package(url: "https://github.com/apple/swift-system", from: "1.5.0"),
+        .package(url: "https://github.com/apple/swift-collections", from: "1.3.0"),
         .package(url: "https://github.com/swift-server/swift-service-lifecycle", from: "2.7.0"),
         .package(url: "https://github.com/apple/swift-log", from: "1.6.3"),
         .package(url: "https://github.com/apple/swift-metrics", from: "2.7.0"),
@@ -92,6 +94,10 @@ let package = Package(
                     name: "SystemPackage",
                     package: "swift-system"
                 ),
+                .product(
+                    name: "DequeModule",
+                    package: "swift-collections"
+                ),
                 .product(
                     name: "Logging",
                     package: "swift-log",
@@ -179,8 +185,16 @@ for target in package.targets {
     // https://github.com/swiftlang/swift-evolution/blob/main/proposals/0409-access-level-on-imports.md
     settings.append(.enableUpcomingFeature("InternalImportsByDefault"))
 
+    // https://docs.swift.org/compiler/documentation/diagnostics/nonisolated-nonsending-by-default/
+    settings.append(.enableUpcomingFeature("NonisolatedNonsendingByDefault"))
+
     settings.append(.enableExperimentalFeature("AvailabilityMacro=Configuration 1.0:macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0"))
 
+    if enableAllCIFlags {
+        // Ensure all public types are explicitly annotated as Sendable or not Sendable.
+        settings.append(.unsafeFlags(["-Xfrontend", "-require-explicit-sendable"]))
+    }
+
     target.swiftSettings = settings
 }
 

Sources/Configuration/ConfigProviderHelpers.swift

@@ -42,13 +42,15 @@ extension ConfigProvider {
     ///   - updatesHandler: The closure that processes the async sequence of value updates.
     /// - Returns: The value returned by the handler closure.
     /// - Throws: Provider-specific errors or errors thrown by the handler.
-    public func watchValueFromValue<Return>(
-        forKey key: AbsoluteConfigKey,
-        type: ConfigType,
-        updatesHandler: (
-            ConfigUpdatesAsyncSequence<Result<LookupResult, any Error>, Never>
+    nonisolated(nonsending)
+        public func watchValueFromValue<Return>(
+            forKey key: AbsoluteConfigKey,
+            type: ConfigType,
+            updatesHandler: (
+                ConfigUpdatesAsyncSequence<Result<LookupResult, any Error>, Never>
+            ) async throws -> Return
         ) async throws -> Return
-    ) async throws -> Return {
+    {
         let (stream, continuation) = AsyncStream<Result<LookupResult, any Error>>
             .makeStream(bufferingPolicy: .bufferingNewest(1))
         let initialValue: Result<LookupResult, any Error>
@@ -83,9 +85,11 @@ extension ConfigProvider {
     /// - Parameter updatesHandler: The closure that processes the async sequence of snapshot updates.
     /// - Returns: The value returned by the handler closure.
     /// - Throws: Provider-specific errors or errors thrown by the handler.
-    public func watchSnapshotFromSnapshot<Return>(
-        updatesHandler: (ConfigUpdatesAsyncSequence<any ConfigSnapshotProtocol, Never>) async throws -> Return
-    ) async throws -> Return {
+    nonisolated(nonsending)
+        public func watchSnapshotFromSnapshot<Return>(
+            updatesHandler: (ConfigUpdatesAsyncSequence<any ConfigSnapshotProtocol, Never>) async throws -> Return
+        ) async throws -> Return
+    {
         let (stream, continuation) = AsyncStream<any ConfigSnapshotProtocol>
             .makeStream(bufferingPolicy: .bufferingNewest(1))
         let initialValue = snapshot()

Sources/Configuration/Documentation.docc/Reference/ConfigReader-Watch.md

@@ -4,47 +4,47 @@
 
 ### Watching string values
 - ``ConfigReader/watchString(forKey:isSecret:fileID:line:updatesHandler:)``
-- ``ConfigReader/watchString(forKey:as:isSecret:fileID:line:updatesHandler:)-4q1c0``
-- ``ConfigReader/watchString(forKey:as:isSecret:fileID:line:updatesHandler:)-7lki4``
+- ``ConfigReader/watchString(forKey:as:isSecret:fileID:line:updatesHandler:)-7mxw1``
+- ``ConfigReader/watchString(forKey:as:isSecret:fileID:line:updatesHandler:)-818sy``
 - ``ConfigReader/watchString(forKey:isSecret:default:fileID:line:updatesHandler:)``
-- ``ConfigReader/watchString(forKey:as:isSecret:default:fileID:line:updatesHandler:)-4x6zt``
-- ``ConfigReader/watchString(forKey:as:isSecret:default:fileID:line:updatesHandler:)-1ncw1``
+- ``ConfigReader/watchString(forKey:as:isSecret:default:fileID:line:updatesHandler:)-6m0yu``
+- ``ConfigReader/watchString(forKey:as:isSecret:default:fileID:line:updatesHandler:)-6dpc3``
 - ``ConfigReader/watchString(forKey:context:isSecret:fileID:line:updatesHandler:)``
-- ``ConfigReader/watchString(forKey:context:as:isSecret:fileID:line:updatesHandler:)-1vua5``
-- ``ConfigReader/watchString(forKey:context:as:isSecret:fileID:line:updatesHandler:)-1s8wu``
+- ``ConfigReader/watchString(forKey:context:as:isSecret:fileID:line:updatesHandler:)-34wbx``
+- ``ConfigReader/watchString(forKey:context:as:isSecret:fileID:line:updatesHandler:)-549xr``
 - ``ConfigReader/watchString(forKey:context:isSecret:default:fileID:line:updatesHandler:)``
-- ``ConfigReader/watchString(forKey:context:as:isSecret:default:fileID:line:updatesHandler:)-3ppdh``
-- ``ConfigReader/watchString(forKey:context:as:isSecret:default:fileID:line:updatesHandler:)-80t2z``
+- ``ConfigReader/watchString(forKey:context:as:isSecret:default:fileID:line:updatesHandler:)-9u7vf``
+- ``ConfigReader/watchString(forKey:context:as:isSecret:default:fileID:line:updatesHandler:)-1ofiv``
 
 ### Watching required string values
 - ``ConfigReader/watchRequiredString(forKey:isSecret:fileID:line:updatesHandler:)``
-- ``ConfigReader/watchRequiredString(forKey:as:isSecret:fileID:line:updatesHandler:)-29xb0``
-- ``ConfigReader/watchRequiredString(forKey:as:isSecret:fileID:line:updatesHandler:)-3dox3``
+- ``ConfigReader/watchRequiredString(forKey:as:isSecret:fileID:line:updatesHandler:)-86ot1``
+- ``ConfigReader/watchRequiredString(forKey:as:isSecret:fileID:line:updatesHandler:)-3lrs7``
 - ``ConfigReader/watchRequiredString(forKey:context:isSecret:fileID:line:updatesHandler:)``
-- ``ConfigReader/watchRequiredString(forKey:context:as:isSecret:fileID:line:updatesHandler:)-6v7w5``
-- ``ConfigReader/watchRequiredString(forKey:context:as:isSecret:fileID:line:updatesHandler:)-76kbb``
+- ``ConfigReader/watchRequiredString(forKey:context:as:isSecret:fileID:line:updatesHandler:)-77978``
+- ``ConfigReader/watchRequiredString(forKey:context:as:isSecret:fileID:line:updatesHandler:)-138o2``
 
 ### Watching lists of string values
 - ``ConfigReader/watchStringArray(forKey:isSecret:fileID:line:updatesHandler:)``
-- ``ConfigReader/watchStringArray(forKey:as:isSecret:fileID:line:updatesHandler:)-5igvu``
-- ``ConfigReader/watchStringArray(forKey:as:isSecret:fileID:line:updatesHandler:)-38ruy``
+- ``ConfigReader/watchStringArray(forKey:as:isSecret:fileID:line:updatesHandler:)-8t4nb``
+- ``ConfigReader/watchStringArray(forKey:as:isSecret:fileID:line:updatesHandler:)-9cmju``
 - ``ConfigReader/watchStringArray(forKey:isSecret:default:fileID:line:updatesHandler:)``
-- ``ConfigReader/watchStringArray(forKey:as:isSecret:default:fileID:line:updatesHandler:)-7oi5b``
-- ``ConfigReader/watchStringArray(forKey:as:isSecret:default:fileID:line:updatesHandler:)-4rhx2``
+- ``ConfigReader/watchStringArray(forKey:as:isSecret:default:fileID:line:updatesHandler:)-59de``
+- ``ConfigReader/watchStringArray(forKey:as:isSecret:default:fileID:line:updatesHandler:)-8nsil``
 - ``ConfigReader/watchStringArray(forKey:context:isSecret:fileID:line:updatesHandler:)``
-- ``ConfigReader/watchStringArray(forKey:context:as:isSecret:fileID:line:updatesHandler:)-6gaip``
-- ``ConfigReader/watchStringArray(forKey:context:as:isSecret:fileID:line:updatesHandler:)-5dyyx``
+- ``ConfigReader/watchStringArray(forKey:context:as:isSecret:fileID:line:updatesHandler:)-5occx``
+- ``ConfigReader/watchStringArray(forKey:context:as:isSecret:fileID:line:updatesHandler:)-30hf0``
 - ``ConfigReader/watchStringArray(forKey:context:isSecret:default:fileID:line:updatesHandler:)``
-- ``ConfigReader/watchStringArray(forKey:context:as:isSecret:default:fileID:line:updatesHandler:)-7tbs9``
-- ``ConfigReader/watchStringArray(forKey:context:as:isSecret:default:fileID:line:updatesHandler:)-5yo2r``
+- ``ConfigReader/watchStringArray(forKey:context:as:isSecret:default:fileID:line:updatesHandler:)-4txm0``
+- ``ConfigReader/watchStringArray(forKey:context:as:isSecret:default:fileID:line:updatesHandler:)-3eipe``
 
 ### Watching required lists of string values
 - ``ConfigReader/watchRequiredStringArray(forKey:isSecret:fileID:line:updatesHandler:)``
-- ``ConfigReader/watchRequiredStringArray(forKey:as:isSecret:fileID:line:updatesHandler:)-1t82o``
-- ``ConfigReader/watchRequiredStringArray(forKey:as:isSecret:fileID:line:updatesHandler:)-7lk1k``
+- ``ConfigReader/watchRequiredStringArray(forKey:as:isSecret:fileID:line:updatesHandler:)-3whiy``
+- ``ConfigReader/watchRequiredStringArray(forKey:as:isSecret:fileID:line:updatesHandler:)-4zyyq``
 - ``ConfigReader/watchRequiredStringArray(forKey:context:isSecret:fileID:line:updatesHandler:)``
-- ``ConfigReader/watchRequiredStringArray(forKey:context:as:isSecret:fileID:line:updatesHandler:)-5zo1e``
-- ``ConfigReader/watchRequiredStringArray(forKey:context:as:isSecret:fileID:line:updatesHandler:)-6kvcj``
+- ``ConfigReader/watchRequiredStringArray(forKey:context:as:isSecret:fileID:line:updatesHandler:)-97r4l``
+- ``ConfigReader/watchRequiredStringArray(forKey:context:as:isSecret:fileID:line:updatesHandler:)-4jcy3``
 
 ### Watching Boolean values
 - ``ConfigReader/watchBool(forKey:isSecret:fileID:line:updatesHandler:)``

Sources/Configuration/MultiProvider.swift

@@ -195,28 +195,33 @@ extension MultiProvider {
     /// - Parameter body: A closure that receives an async sequence of ``MultiSnapshot`` updates.
     /// - Returns: The value returned by the body closure.
     /// - Throws: Any error thrown by the nested providers or the body closure.
-    func watchSnapshot<Return>(
-        _ body: (ConfigUpdatesAsyncSequence<MultiSnapshot, Never>) async throws -> Return
-    ) async throws -> Return {
+    nonisolated(nonsending)
+        func watchSnapshot<Return>(
+            _ body: (ConfigUpdatesAsyncSequence<MultiSnapshot, Never>) async throws -> Return
+        ) async throws -> Return
+    {
         let providers = storage.providers
-        let sources:
-            [@Sendable (
-                (ConfigUpdatesAsyncSequence<any ConfigSnapshotProtocol, Never>) async throws -> Void
-            ) async throws -> Void] = providers.map { $0.watchSnapshot }
-        return try await combineLatestOneOrMore(
-            elementType: (any ConfigSnapshotProtocol).self,
-            sources: sources,
-            updatesHandler: { updateArrays in
-                try await body(
-                    ConfigUpdatesAsyncSequence(
-                        updateArrays
-                            .map { array in
-                                MultiSnapshot(snapshots: array)
-                            }
-                    )
+        typealias UpdatesSequence = any (AsyncSequence<any ConfigSnapshotProtocol, Never> & Sendable)
+        var updateSequences: [UpdatesSequence] = []
+        updateSequences.reserveCapacity(providers.count)
+        return try await withProvidersWatchingSnapshot(
+            providers: ArraySlice(providers),
+            updateSequences: &updateSequences,
+        ) { providerUpdateSequences in
+            let updateArrays = combineLatestMany(
+                elementType: (any ConfigSnapshotProtocol).self,
+                failureType: Never.self,
+                providerUpdateSequences
+            )
+            return try await body(
+                ConfigUpdatesAsyncSequence(
+                    updateArrays
+                        .map { array in
+                            MultiSnapshot(snapshots: array)
+                        }
                 )
-            }
-        )
+            )
+        }
     }
 
     /// Asynchronously resolves a configuration value from nested providers.
@@ -281,52 +286,97 @@ extension MultiProvider {
     ///   - updatesHandler: A closure that receives an async sequence of combined updates from all providers.
     /// - Throws: Any error thrown by the nested providers or the handler closure.
     /// - Returns: The value returned by the handler.
-    func watchValue<Return>(
-        forKey key: AbsoluteConfigKey,
-        type: ConfigType,
-        updatesHandler: (
-            ConfigUpdatesAsyncSequence<([AccessEvent.ProviderResult], Result<ConfigValue?, any Error>), Never>
+    nonisolated(nonsending)
+        func watchValue<Return>(
+            forKey key: AbsoluteConfigKey,
+            type: ConfigType,
+            updatesHandler: (
+                ConfigUpdatesAsyncSequence<([AccessEvent.ProviderResult], Result<ConfigValue?, any Error>), Never>
+            ) async throws -> Return
         ) async throws -> Return
-    ) async throws -> Return {
+    {
         let providers = storage.providers
         let providerNames = providers.map(\.providerName)
-        let sources:
-            [@Sendable (
-                (
-                    ConfigUpdatesAsyncSequence<Result<LookupResult, any Error>, Never>
-                ) async throws -> Void
-            ) async throws -> Void] = providers.map { provider in
-                { handler in
-                    _ = try await provider.watchValue(forKey: key, type: type, updatesHandler: handler)
-                }
-            }
-        return try await combineLatestOneOrMore(
-            elementType: Result<LookupResult, any Error>.self,
-            sources: sources,
-            updatesHandler: { updateArrays in
-                try await updatesHandler(
-                    ConfigUpdatesAsyncSequence(
-                        updateArrays
-                            .map { array in
-                                var results: [AccessEvent.ProviderResult] = []
-                                for (providerIndex, lookupResult) in array.enumerated() {
-                                    let providerName = providerNames[providerIndex]
-                                    results.append(.init(providerName: providerName, result: lookupResult))
-                                    switch lookupResult {
-                                    case .success(let value) where value.value == nil:
-                                        // Got a success + nil from a nested provider, keep iterating.
-                                        continue
-                                    default:
-                                        // Got a success + non-nil or an error from a nested provider, propagate that up.
-                                        return (results, lookupResult.map { $0.value })
-                                    }
+        typealias UpdatesSequence = any (AsyncSequence<Result<LookupResult, any Error>, Never> & Sendable)
+        var updateSequences: [UpdatesSequence] = []
+        updateSequences.reserveCapacity(providers.count)
+        return try await withProvidersWatchingValue(
+            providers: ArraySlice(providers),
+            updateSequences: &updateSequences,
+            key: key,
+            configType: type,
+        ) { providerUpdateSequences in
+            let updateArrays = combineLatestMany(
+                elementType: Result<LookupResult, any Error>.self,
+                failureType: Never.self,
+                providerUpdateSequences
+            )
+            return try await updatesHandler(
+                ConfigUpdatesAsyncSequence(
+                    updateArrays
+                        .map { array in
+                            var results: [AccessEvent.ProviderResult] = []
+                            for (providerIndex, lookupResult) in array.enumerated() {
+                                let providerName = providerNames[providerIndex]
+                                results.append(.init(providerName: providerName, result: lookupResult))
+                                switch lookupResult {
+                                case .success(let value) where value.value == nil:
+                                    // Got a success + nil from a nested provider, keep iterating.
+                                    continue
+                                default:
+                                    // Got a success + non-nil or an error from a nested provider, propagate that up.
+                                    return (results, lookupResult.map { $0.value })
                                 }
-                                // If all nested results were success + nil, return the same.
-                                return (results, .success(nil))
                             }
-                    )
+                            // If all nested results were success + nil, return the same.
+                            return (results, .success(nil))
+                        }
                 )
-            }
+            )
+        }
+    }
+}
+
+@available(Configuration 1.0, *)
+nonisolated(nonsending) private func withProvidersWatchingValue<ReturnInner>(
+    providers: ArraySlice<any ConfigProvider>,
+    updateSequences: inout [any (AsyncSequence<Result<LookupResult, any Error>, Never> & Sendable)],
+    key: AbsoluteConfigKey,
+    configType: ConfigType,
+    body: ([any (AsyncSequence<Result<LookupResult, any Error>, Never> & Sendable)]) async throws -> ReturnInner
+) async throws -> ReturnInner {
+    guard let provider = providers.first else {
+        // Recursion termination, once we've collected all update sequences, execute the body.
+        return try await body(updateSequences)
+    }
+    return try await provider.watchValue(forKey: key, type: configType) { updates in
+        updateSequences.append(updates)
+        return try await withProvidersWatchingValue(
+            providers: providers.dropFirst(),
+            updateSequences: &updateSequences,
+            key: key,
+            configType: configType,
+            body: body
+        )
+    }
+}
+
+@available(Configuration 1.0, *)
+nonisolated(nonsending) private func withProvidersWatchingSnapshot<ReturnInner>(
+    providers: ArraySlice<any ConfigProvider>,
+    updateSequences: inout [any (AsyncSequence<any ConfigSnapshotProtocol, Never> & Sendable)],
+    body: ([any (AsyncSequence<any ConfigSnapshotProtocol, Never> & Sendable)]) async throws -> ReturnInner
+) async throws -> ReturnInner {
+    guard let provider = providers.first else {
+        // Recursion termination, once we've collected all update sequences, execute the body.
+        return try await body(updateSequences)
+    }
+    return try await provider.watchSnapshot { updates in
+        updateSequences.append(updates)
+        return try await withProvidersWatchingSnapshot(
+            providers: providers.dropFirst(),
+            updateSequences: &updateSequences,
+            body: body
         )
     }
 }