Deprecate sync methods in RegistryDownloadsManager (#8323)

plemarquand · web-flow · commit 2d03ce7a858e · 2025-03-05T09:30:19.000-05:00
Deprecate the synchronous methods in RegistryDownloadsManager and move the logic from the existing sync methods to the async methods. The synchronous methods in RegistryDownloadsManager will now warn when used. After some time they can be removed completely. This patch follows the same refactoring conventions as #8263
diff --git a/Sources/Basics/FileSystem/FileSystem+Extensions.swift b/Sources/Basics/FileSystem/FileSystem+Extensions.swift
@@ -199,6 +199,11 @@ extension FileSystem {
         try self.withLock(on: path.underlying, type: type, blocking: blocking, body)
     }
 
+    /// Execute the given block while holding the lock.
+    public func withLock<T>(on path: AbsolutePath, type: FileLock.LockType, blocking: Bool = true, _ body: () async throws -> T) async throws -> T {
+        return try await FileLock.withLock(fileToLock: path.underlying, type: type, blocking: blocking, body: body)
+    }
+
     /// Returns any known item replacement directories for a given path. These may be used by platform-specific
     /// libraries to handle atomic file system operations, such as deletion.
     func itemReplacementDirectories(for path: AbsolutePath) throws -> [AbsolutePath] {
diff --git a/Sources/PackageRegistry/RegistryDownloadsManager.swift b/Sources/PackageRegistry/RegistryDownloadsManager.swift
@@ -28,7 +28,12 @@ public class RegistryDownloadsManager: AsyncCancellable {
     private let registryClient: RegistryClient
     private let delegate: Delegate?
 
-    private var pendingLookups = [PackageIdentity: DispatchGroup]()
+    struct PackageLookup: Hashable {
+        let package: PackageIdentity
+        let version: Version
+    }
+
+    private var pendingLookups = [PackageLookup: Task<AbsolutePath, Error>]()
     private var pendingLookupsLock = NSLock()
 
     public init(
@@ -44,24 +49,77 @@ public class RegistryDownloadsManager: AsyncCancellable {
         self.registryClient = registryClient
         self.delegate = delegate
     }
-    
+
     public func lookup(
         package: PackageIdentity,
         version: Version,
         observabilityScope: ObservabilityScope,
         delegateQueue: DispatchQueue,
         callbackQueue: DispatchQueue
     ) async throws -> AbsolutePath {
-        try await withCheckedThrowingContinuation { continuation in
-            self.lookup(
-                package: package,
-                version: version,
-                observabilityScope: observabilityScope,
-                delegateQueue: delegateQueue,
-                callbackQueue: callbackQueue,
-                completion: { continuation.resume(with: $0) }
-            )
+        let packageRelativePath: RelativePath
+        let packagePath: AbsolutePath
+
+        packageRelativePath = try package.downloadPath(version: version)
+        packagePath = self.path.appending(packageRelativePath)
+
+        // TODO: we can do some finger-print checking to improve the validation
+        // already exists and valid, we can exit early
+        if try self.fileSystem.validPackageDirectory(packagePath) {
+            return packagePath
         }
+
+        let lookupId = PackageLookup(package: package, version: version)
+        let task = await withCheckedContinuation { continuation in
+            self.pendingLookupsLock.lock()
+            defer { self.pendingLookupsLock.unlock() }
+
+            // Check if we've already resolved/are in the process of resolving for this package.
+            if let inFlight = self.pendingLookups[lookupId] {
+                continuation.resume(returning: inFlight)
+            } else {
+                let lookupTask = Task {
+                    // inform delegate that we are starting to fetch
+                    // calculate if cached (for delegate call) outside queue as it may change while queue is processing
+                    let isCached = self.cachePath.map { self.fileSystem.exists($0.appending(packageRelativePath)) } ?? false
+                    delegateQueue.async { [delegate = self.delegate] in
+                        let details = FetchDetails(fromCache: isCached, updatedCache: false)
+                        delegate?.willFetch(package: package, version: version, fetchDetails: details)
+                    }
+
+                    // make sure destination is free.
+                    try? self.fileSystem.removeFileTree(packagePath)
+
+                    let start = DispatchTime.now()
+                    do {
+                        let result = try await self.downloadAndPopulateCache(
+                            package: package,
+                            version: version,
+                            packagePath: packagePath,
+                            observabilityScope: observabilityScope,
+                            delegateQueue: delegateQueue,
+                            callbackQueue: callbackQueue,
+                        )
+                        // inform delegate that we finished to fetch
+                        let duration = start.distance(to: .now())
+                        delegateQueue.async { [delegate = self.delegate] in
+                            delegate?.didFetch(package: package, version: version, result: .success(result), duration: duration)
+                        }
+                    } catch {
+                        let duration = start.distance(to: .now())
+                        delegateQueue.async { [delegate = self.delegate] in
+                            delegate?.didFetch(package: package, version: version, result: .failure(error), duration: duration)
+                        }
+                        throw error
+                    }
+                    return packagePath
+                }
+
+                self.pendingLookups[lookupId] = lookupTask
+                continuation.resume(returning: lookupTask)
+            }
+        }
+        return try await task.value
     }
 
     @available(*, noasync, message: "Use the async alternative")
@@ -73,82 +131,14 @@ public class RegistryDownloadsManager: AsyncCancellable {
         callbackQueue: DispatchQueue,
         completion: @escaping (Result<AbsolutePath, Error>) -> Void
     ) {
-        // wrap the callback in the requested queue
-        let completion = { result in callbackQueue.async { completion(result) } }
-
-        let packageRelativePath: RelativePath
-        let packagePath: AbsolutePath
-
-        do {
-            packageRelativePath = try package.downloadPath(version: version)
-            packagePath = self.path.appending(packageRelativePath)
-
-            // TODO: we can do some finger-print checking to improve the validation
-            // already exists and valid, we can exit early
-            if try self.fileSystem.validPackageDirectory(packagePath) {
-                return completion(.success(packagePath))
-            }
-        } catch {
-            return completion(.failure(error))
-        }
-
-        // next we check if there is a pending lookup
-        self.pendingLookupsLock.lock()
-        if let pendingLookup = self.pendingLookups[package] {
-            self.pendingLookupsLock.unlock()
-            // chain onto the pending lookup
-            pendingLookup.notify(queue: callbackQueue) {
-                // at this point the previous lookup should be complete and we can re-lookup
-                self.lookup(
-                    package: package,
-                    version: version,
-                    observabilityScope: observabilityScope,
-                    delegateQueue: delegateQueue,
-                    callbackQueue: callbackQueue,
-                    completion: completion
-                )
-            }
-        } else {
-            // record the pending lookup
-            assert(self.pendingLookups[package] == nil)
-            let group = DispatchGroup()
-            group.enter()
-            self.pendingLookups[package] = group
-            self.pendingLookupsLock.unlock()
-
-            // inform delegate that we are starting to fetch
-            // calculate if cached (for delegate call) outside queue as it may change while queue is processing
-            let isCached = self.cachePath.map { self.fileSystem.exists($0.appending(packageRelativePath)) } ?? false
-            delegateQueue.async {
-                let details = FetchDetails(fromCache: isCached, updatedCache: false)
-                self.delegate?.willFetch(package: package, version: version, fetchDetails: details)
-            }
-
-            // make sure destination is free.
-            try? self.fileSystem.removeFileTree(packagePath)
-
-            let start = DispatchTime.now()
-            self.downloadAndPopulateCache(
+        self.executeAsync(completion, on: callbackQueue) {
+            try await self.lookup(
                 package: package,
                 version: version,
-                packagePath: packagePath,
                 observabilityScope: observabilityScope,
                 delegateQueue: delegateQueue,
-                callbackQueue: callbackQueue
-            ) { result in
-                // inform delegate that we finished to fetch
-                let duration = start.distance(to: .now())
-                delegateQueue.async {
-                    self.delegate?.didFetch(package: package, version: version, result: result, duration: duration)
-                }
-                // remove the pending lookup
-                self.pendingLookupsLock.lock()
-                self.pendingLookups[package]?.leave()
-                self.pendingLookups[package] = nil
-                self.pendingLookupsLock.unlock()
-                // and done
-                completion(result.map { _ in packagePath })
-            }
+                callbackQueue: callbackQueue,
+            )
         }
     }
 
@@ -164,15 +154,15 @@ public class RegistryDownloadsManager: AsyncCancellable {
         observabilityScope: ObservabilityScope,
         delegateQueue: DispatchQueue,
         callbackQueue: DispatchQueue,
-        completion: @escaping @Sendable (Result<FetchDetails, Error>) -> Void
-    ) {
+    ) async throws -> FetchDetails {
         if let cachePath {
             do {
                 let relativePath = try package.downloadPath(version: version)
                 let cachedPackagePath = cachePath.appending(relativePath)
 
                 try self.initializeCacheIfNeeded(cachePath: cachePath)
-                try self.fileSystem.withLock(on: cachedPackagePath, type: .exclusive) {
+
+                return try await self.fileSystem.withLock(on: cachedPackagePath, type: .exclusive) {
                     // download the package into the cache unless already exists
                     if try self.fileSystem.validPackageDirectory(cachedPackagePath) {
                         // extra validation to defend from racy edge cases
@@ -182,33 +172,39 @@ public class RegistryDownloadsManager: AsyncCancellable {
                         // copy the package from the cache into the package path.
                         try self.fileSystem.createDirectory(packagePath.parentDirectory, recursive: true)
                         try self.fileSystem.copy(from: cachedPackagePath, to: packagePath)
-                        completion(.success(.init(fromCache: true, updatedCache: false)))
+                        return FetchDetails(fromCache: true, updatedCache: false)
                     } else {
-                        // it is possible that we already created the directory before from failed attempts, so clear leftover data if present.
-                        try? self.fileSystem.removeFileTree(cachedPackagePath)
-                        // download the package from the registry
-                        self.registryClient.downloadSourceArchive(
-                            package: package,
-                            version: version,
-                            destinationPath: cachedPackagePath,
-                            progressHandler: updateDownloadProgress,
-                            fileSystem: self.fileSystem,
-                            observabilityScope: observabilityScope,
-                            callbackQueue: callbackQueue
-                        ) { result in
-                            completion(result.tryMap {
-                                // extra validation to defend from racy edge cases
-                                if self.fileSystem.exists(packagePath) {
-                                    throw StringError("\(packagePath) already exists unexpectedly")
-                                }
-                                // copy the package from the cache into the package path.
-                                try self.fileSystem.createDirectory(packagePath.parentDirectory, recursive: true)
-                                try self.fileSystem.copy(from: cachedPackagePath, to: packagePath)
-                                return FetchDetails(fromCache: true, updatedCache: true)
-                            })
+                        do {
+                            // it is possible that we already created the directory before from failed attempts, so clear leftover data if present.
+                            try? self.fileSystem.removeFileTree(cachedPackagePath)
+                            // download the package from the registry
+                            let _ = try await self.registryClient.downloadSourceArchive(
+                                package: package,
+                                version: version,
+                                destinationPath: cachedPackagePath,
+                                progressHandler: updateDownloadProgress,
+                                fileSystem: self.fileSystem,
+                                observabilityScope: observabilityScope,
+                                callbackQueue: callbackQueue
+                            )
+
+                            // extra validation to defend from racy edge cases
+                            if self.fileSystem.exists(packagePath) {
+                                throw StringError("\(packagePath) already exists unexpectedly")
+                            }
+                            // copy the package from the cache into the package path.
+                            try self.fileSystem.createDirectory(packagePath.parentDirectory, recursive: true)
+                            try self.fileSystem.copy(from: cachedPackagePath, to: packagePath)
+                            return FetchDetails(fromCache: true, updatedCache: true)
+                        } catch {
+                            // Wrap this error to do a straight rethrow instead of handling it as if
+                            // the download should be made without populating the cache.
+                            throw DownloadError.passthrough(error)
                         }
                     }
                 }
+            } catch DownloadError.passthrough(let underlyingError) {
+                throw underlyingError
             } catch {
                 // download without populating the cache in the case of an error.
                 observabilityScope.emit(
@@ -217,52 +213,56 @@ public class RegistryDownloadsManager: AsyncCancellable {
                 )
                 // it is possible that we already created the directory from failed attempts, so clear leftover data if present.
                 try? self.fileSystem.removeFileTree(packagePath)
-                self.registryClient.downloadSourceArchive(
+                let _ = try await self.registryClient.downloadSourceArchive(
                     package: package,
                     version: version,
                     destinationPath: packagePath,
                     progressHandler: updateDownloadProgress,
                     fileSystem: self.fileSystem,
                     observabilityScope: observabilityScope,
                     callbackQueue: callbackQueue
-                ) { result in
-                    completion(result.map { FetchDetails(fromCache: false, updatedCache: false) })
-                }
+                )
+                return FetchDetails(fromCache: false, updatedCache: false)
             }
         } else {
             // it is possible that we already created the directory from failed attempts, so clear leftover data if present.
             try? self.fileSystem.removeFileTree(packagePath)
+
             // download without populating the cache when no `cachePath` is set.
-            self.registryClient.downloadSourceArchive(
+            let _ = try await self.registryClient.downloadSourceArchive(
                 package: package,
                 version: version,
                 destinationPath: packagePath,
                 progressHandler: updateDownloadProgress,
                 fileSystem: self.fileSystem,
                 observabilityScope: observabilityScope,
                 callbackQueue: callbackQueue
-            ) { result in
-                completion(result.map { FetchDetails(fromCache: false, updatedCache: false) })
-            }
+            )
+            return FetchDetails(fromCache: false, updatedCache: false)
         }
 
         // utility to update progress
 
         @Sendable func updateDownloadProgress(downloaded: Int64, total: Int64?) {
-            delegateQueue.async {
-                self.delegate?.fetching(
+            delegateQueue.async { [delegate = self.delegate] in
+                delegate?.fetching(
                     package: package,
                     version: version,
                     bytesDownloaded: downloaded,
                     totalBytesToDownload: total
                 )
             }
         }
+
+        enum DownloadError: Error {
+            case passthrough(Error)
+        }
     }
 
     public func remove(package: PackageIdentity) throws {
         let relativePath = try package.downloadPath()
         let packagesPath = self.path.appending(relativePath)
+        self.pendingLookups.removeValue(forPackage: package)
         try self.fileSystem.removeFileTree(packagesPath)
     }
 
@@ -314,10 +314,25 @@ public class RegistryDownloadsManager: AsyncCancellable {
             try self.fileSystem.createDirectory(cachePath, recursive: true)
         }
     }
+
+    private func executeAsync<T>(
+        _ callback: @escaping (Result<T, Error>) -> Void,
+        on queue: DispatchQueue,
+        _ closure: @escaping () async throws -> T
+    ) {
+        let completion: (Result<T, Error>) -> Void = { result in queue.async { callback(result) } }
+        Task {
+            do {
+                completion(.success(try await closure()))
+            } catch {
+                completion(.failure(error))
+            }
+        }
+    }
 }
 
 /// Delegate to notify clients about actions being performed by RegistryManager.
-public protocol RegistryDownloadsManagerDelegate {
+public protocol RegistryDownloadsManagerDelegate: Sendable {
     /// Called when a package is about to be fetched.
     func willFetch(package: PackageIdentity, version: Version, fetchDetails: RegistryDownloadsManager.FetchDetails)
 
@@ -333,9 +348,17 @@ public protocol RegistryDownloadsManagerDelegate {
     func fetching(package: PackageIdentity, version: Version, bytesDownloaded: Int64, totalBytesToDownload: Int64?)
 }
 
+extension Dictionary where Key == RegistryDownloadsManager.PackageLookup {
+    fileprivate mutating func removeValue(forPackage package: PackageIdentity) {
+        self.keys
+            .filter { $0.package == package }
+            .forEach { self.removeValue(forKey: $0) }
+    }
+}
+
 extension RegistryDownloadsManager {
     /// Additional information about a fetch
-    public struct FetchDetails: Equatable {
+    public struct FetchDetails: Equatable, Sendable {
         /// Indicates if the repository was fetched from the cache or from the remote.
         public let fromCache: Bool
         /// Indicates whether the repository was already present in the cache and updated or if a clean fetch was performed.
diff --git a/Tests/PackageRegistryTests/RegistryDownloadsManagerTests.swift b/Tests/PackageRegistryTests/RegistryDownloadsManagerTests.swift
