Deprecate sync methods in RegistryClient (#8263)

Deprecate the synchronous methods in RegistryClient and move the logic
from then sync versions in to the async versions. This async-ifys the
previously callback based methods, flattening them out.

The synchronous versions of these methods will now warn when they're
used. After some time they can be removed completely.

Author: plemarquand

Sources/Basics/Cancellator.swift

@@ -19,18 +19,13 @@ import WinSDK
 import Android
 #endif
 
-public typealias CancellationHandler = @Sendable (DispatchTime) throws -> Void
+public typealias CancellationHandler = @Sendable (DispatchTime) async throws -> Void
 
 public final class Cancellator: Cancellable, Sendable {
     public typealias RegistrationKey = String
 
     private let observabilityScope: ObservabilityScope?
     private let registry = ThreadSafeKeyValueStore<String, (name: String, handler: CancellationHandler)>()
-    private let cancelationQueue = DispatchQueue(
-        label: "org.swift.swiftpm.cancellator",
-        qos: .userInteractive,
-        attributes: .concurrent
-    )
     private let cancelling = ThreadSafeBox<Bool>(false)
 
     private static let signalHandlerLock = NSLock()
@@ -119,6 +114,11 @@ public final class Cancellator: Cancellable, Sendable {
         self.register(name: name, handler: handler.cancel(deadline:))
     }
 
+    @discardableResult
+    public func register(name: String, handler: AsyncCancellable) -> RegistrationKey? {
+        self.register(name: name, handler: handler.cancel(deadline:))
+    }
+
     @discardableResult
     public func register(name: String, handler: @escaping @Sendable () throws -> Void) -> RegistrationKey? {
         self.register(name: name, handler: { _ in try handler() })
@@ -159,10 +159,12 @@ public final class Cancellator: Cancellable, Sendable {
         let cancelled = ThreadSafeArrayStore<String>()
         let group = DispatchGroup()
         for (_, (name, handler)) in cancellationHandlers {
-            self.cancelationQueue.async(group: group) {
+            group.enter()
+            Task {
+                defer { group.leave() }
                 do {
                     self.observabilityScope?.emit(debug: "cancelling '\(name)'")
-                    try handler(handlersDeadline)
+                    try await handler(handlersDeadline)
                     cancelled.append(name)
                 } catch {
                     self.observabilityScope?.emit(
@@ -192,6 +194,10 @@ public protocol Cancellable {
     func cancel(deadline: DispatchTime) throws -> Void
 }
 
+public protocol AsyncCancellable {
+    func cancel(deadline: DispatchTime) async throws -> Void
+}
+
 public struct CancellationError: Error, CustomStringConvertible {
     public let description: String
 

Sources/Basics/HTTPClient/HTTPClient.swift

@@ -40,6 +40,9 @@ public actor HTTPClient {
     /// Array of `HostErrors` values, which is used for applying a circuit-breaking strategy.
     private var hostsErrors = [String: HostErrors]()
 
+    /// Tracks all active network request tasks.
+    private var activeTasks: Set<Task<HTTPClient.Response, Error>> = []
+
     public init(configuration: HTTPClientConfiguration = .init(), implementation: Implementation? = nil) {
         self.configuration = configuration
         self.implementation = implementation ?? URLSessionHTTPClient().execute
@@ -92,6 +95,21 @@ public actor HTTPClient {
         return try await self.executeWithStrategies(request: request, requestNumber: 0, observabilityScope, progress)
     }
 
+    /// Cancel all in flight network reqeusts.
+    public func cancel(deadline: DispatchTime) async {
+        for task in activeTasks {
+            task.cancel()
+        }
+
+        // Wait for tasks to complete or timeout
+        while !activeTasks.isEmpty && (deadline.distance(to: .now()).nanoseconds() ?? 0) > 0 {
+            await Task.yield()
+        }
+
+        // Clear out the active task list regardless of whether they completed or not
+        activeTasks.removeAll()
+    }
+
     private func executeWithStrategies(
         request: Request,
         requestNumber: Int,
@@ -104,46 +122,57 @@ public actor HTTPClient {
             throw HTTPClientError.circuitBreakerTriggered
         }
 
-        let response = try await self.tokenBucket.withToken {
-            try await self.implementation(request) { received, expected in
-                if let max = request.options.maximumResponseSizeInBytes {
-                    guard received < max else {
-                        // It's a responsibility of the underlying client implementation to cancel the request
-                        // when this closure throws an error
-                        throw HTTPClientError.responseTooLarge(received)
+        let task = Task {
+            let response = try await self.tokenBucket.withToken {
+                try Task.checkCancellation()
+
+                return try await self.implementation(request) { received, expected in
+                    if let max = request.options.maximumResponseSizeInBytes {
+                        guard received < max else {
+                            // It's a responsibility of the underlying client implementation to cancel the request
+                            // when this closure throws an error
+                            throw HTTPClientError.responseTooLarge(received)
+                        }
                     }
-                }
 
-                try progress?(received, expected)
+                    try progress?(received, expected)
+                }
             }
-        }
 
-        self.recordErrorIfNecessary(response: response, request: request)
+            self.recordErrorIfNecessary(response: response, request: request)
 
-        // handle retry strategy
-        if let retryDelay = self.calculateRetry(
-            response: response,
-            request: request,
-            requestNumber: requestNumber
-        ), let retryDelayInNanoseconds = retryDelay.nanoseconds() {
-            observabilityScope?.emit(warning: "\(request.url) failed, retrying in \(retryDelay)")
-            try await Task.sleep(nanoseconds: UInt64(retryDelayInNanoseconds))
-
-            return try await self.executeWithStrategies(
+            // handle retry strategy
+            if let retryDelay = self.calculateRetry(
+                response: response,
                 request: request,
-                requestNumber: requestNumber + 1,
-                observabilityScope,
-                progress
-            )
-        }
-        // check for valid response codes
-        if let validResponseCodes = request.options.validResponseCodes,
-           !validResponseCodes.contains(response.statusCode)
-        {
-            throw HTTPClientError.badResponseStatusCode(response.statusCode)
-        } else {
-            return response
+                requestNumber: requestNumber
+            ), let retryDelayInNanoseconds = retryDelay.nanoseconds() {
+                try Task.checkCancellation()
+
+                observabilityScope?.emit(warning: "\(request.url) failed, retrying in \(retryDelay)")
+                try await Task.sleep(nanoseconds: UInt64(retryDelayInNanoseconds))
+
+                return try await self.executeWithStrategies(
+                    request: request,
+                    requestNumber: requestNumber + 1,
+                    observabilityScope,
+                    progress
+                )
+            }
+            // check for valid response codes
+            if let validResponseCodes = request.options.validResponseCodes,
+            !validResponseCodes.contains(response.statusCode)
+            {
+                throw HTTPClientError.badResponseStatusCode(response.statusCode)
+            } else {
+                return response
+            }
         }
+
+        activeTasks.insert(task)
+        defer { activeTasks.remove(task) }
+
+        return try await task.value
     }
 
     private func calculateRetry(response: Response, request: Request, requestNumber: Int) -> SendableTimeInterval? {
@@ -258,4 +287,26 @@ extension HTTPClient {
             Request(method: .delete, url: url, headers: headers, body: nil, options: options)
         )
     }
+
+    public func download(
+        _ url: URL,
+        headers: HTTPClientHeaders = .init(),
+        options: Request.Options = .init(),
+        progressHandler: ProgressHandler? = nil,
+        fileSystem: FileSystem,
+        destination: AbsolutePath,
+        observabilityScope: ObservabilityScope? = .none
+    ) async throws -> Response {
+        try await self.execute(
+            Request(
+                kind: .download(fileSystem: fileSystem, destination: destination),
+                url: url,
+                headers: headers,
+                body: nil,
+                options: options
+            ),
+            observabilityScope: observabilityScope,
+            progress: progressHandler
+        )
+    }
 }

Sources/PackageRegistry/ChecksumTOFU.swift

@@ -154,7 +154,6 @@ struct PackageVersionChecksumTOFU {
         }
     }
 
-    @available(*, noasync, message: "Use the async alternative")
     func validateManifest(
         registry: Registry,
         package: PackageIdentity.RegistryIdentity,