Cleanup Download Queue

- cleanup various cases conditional on the download queue to always
  use it, now it's default and has been for a while
- ensure that both casks and formulae are fetched at the same time
  when possible to maximise concurrency
- generally DRY up the relevant download queue code

Author: MikeMcQuaid

Library/Homebrew/api.rb

@@ -8,6 +8,7 @@
 require "api/formula_struct"
 require "api/cask_struct"
 require "base64"
+require "download_queue"
 require "utils/output"
 
 module Homebrew
@@ -53,11 +54,13 @@ def self.skip_download?(target:, stale_seconds:)
         endpoint:       String,
         target:         Pathname,
         stale_seconds:  T.nilable(Integer),
-        download_queue: T.nilable(DownloadQueue),
+        download_queue: DownloadQueue,
+        enqueue:        T::Boolean,
       ).returns([T.any(T::Array[T.untyped], T::Hash[String, T.untyped]), T::Boolean])
     }
     def self.fetch_json_api_file(endpoint, target: HOMEBREW_CACHE_API/endpoint,
-                                 stale_seconds: nil, download_queue: nil)
+                                 stale_seconds: nil, download_queue: Homebrew.default_download_queue,
+                                 enqueue: false)
       # Lazy-load dependency.
       require "development_tools"
 
@@ -80,7 +83,7 @@ def self.fetch_json_api_file(endpoint, target: HOMEBREW_CACHE_API/endpoint,
                           DevelopmentTools.curl_substitution_required?
       skip_download = skip_download?(target:, stale_seconds:)
 
-      if download_queue
+      if enqueue
         unless skip_download
           require "api/json_download"
           download = Homebrew::API::JSONDownload.new(endpoint, target:, stale_seconds:)
@@ -167,10 +170,7 @@ def self.merge_variations(json, bottle_tag: T.unsafe(nil))
 
     sig { void }
     def self.fetch_api_files!
-      download_queue = if Homebrew::EnvConfig.download_concurrency > 1
-        require "download_queue"
-        Homebrew::DownloadQueue.new
-      end
+      download_queue = Homebrew::DownloadQueue.new
 
       stale_seconds = if ENV["HOMEBREW_API_UPDATED"].present? ||
                          (Homebrew::EnvConfig.no_auto_update? && !Homebrew::EnvConfig.force_api_auto_update?)
@@ -182,19 +182,19 @@ def self.fetch_api_files!
       end
 
       if Homebrew::EnvConfig.use_internal_api?
-        Homebrew::API::Internal.fetch_formula_api!(download_queue:, stale_seconds:)
-        Homebrew::API::Internal.fetch_cask_api!(download_queue:, stale_seconds:)
+        Homebrew::API::Internal.fetch_formula_api!(download_queue:, stale_seconds:, enqueue: true)
+        Homebrew::API::Internal.fetch_cask_api!(download_queue:, stale_seconds:, enqueue: true)
       else
-        Homebrew::API::Formula.fetch_api!(download_queue:, stale_seconds:)
-        Homebrew::API::Formula.fetch_tap_migrations!(download_queue:, stale_seconds: DEFAULT_API_STALE_SECONDS)
-        Homebrew::API::Cask.fetch_api!(download_queue:, stale_seconds:)
-        Homebrew::API::Cask.fetch_tap_migrations!(download_queue:, stale_seconds: DEFAULT_API_STALE_SECONDS)
+        Homebrew::API::Formula.fetch_api!(download_queue:, stale_seconds:, enqueue: true)
+        Homebrew::API::Formula.fetch_tap_migrations!(download_queue:, stale_seconds: DEFAULT_API_STALE_SECONDS,
+                                                     enqueue: true)
+        Homebrew::API::Cask.fetch_api!(download_queue:, stale_seconds:, enqueue: true)
+        Homebrew::API::Cask.fetch_tap_migrations!(download_queue:, stale_seconds: DEFAULT_API_STALE_SECONDS,
+                                                  enqueue: true)
       end
 
       ENV["HOMEBREW_API_UPDATED"] = "1"
 
-      return unless download_queue
-
       begin
         download_queue.fetch
       ensure

Library/Homebrew/api/cask.rb

@@ -23,20 +23,25 @@ def self.cask_json(name)
         cache.fetch("cask_json").fetch(name)
       end
 
-      sig { params(name: String, download_queue: T.nilable(DownloadQueue)).void }
-      def self.fetch_cask_json!(name, download_queue: nil)
+      sig { params(name: String).void }
+      def self.fetch_cask_json!(name)
         endpoint = "cask/#{name}.json"
-        json_cask, updated = Homebrew::API.fetch_json_api_file endpoint, download_queue: download_queue
-        return if download_queue
+        json_cask, updated = Homebrew::API.fetch_json_api_file endpoint
 
         json_cask = JSON.parse((HOMEBREW_CACHE_API/endpoint).read) unless updated
 
         cache["cask_json"] ||= {}
         cache["cask_json"][name] = json_cask
       end
 
-      sig { params(cask: ::Cask::Cask, download_queue: T.nilable(Homebrew::DownloadQueue)).returns(Homebrew::API::SourceDownload) }
-      def self.source_download(cask, download_queue: nil)
+      sig {
+        params(
+          cask:           ::Cask::Cask,
+          download_queue: Homebrew::DownloadQueue,
+          enqueue:        T::Boolean,
+        ).returns(Homebrew::API::SourceDownload)
+      }
+      def self.source_download(cask, download_queue: Homebrew.default_download_queue, enqueue: false)
         path = cask.ruby_source_path.to_s
         sha256 = cask.ruby_source_checksum[:sha256]
         checksum = Checksum.new(sha256) if sha256
@@ -52,7 +57,7 @@ def self.source_download(cask, download_queue: nil)
           cache:   HOMEBREW_CACHE_API_SOURCE/"#{tap}/#{git_head}/Cask",
         )
 
-        if download_queue
+        if enqueue
           download_queue.enqueue(download)
         elsif !download.symlink_location.exist?
           download.fetch
@@ -75,19 +80,20 @@ def self.cached_json_file_path
       end
 
       sig {
-        params(download_queue: T.nilable(::Homebrew::DownloadQueue), stale_seconds: T.nilable(Integer))
+        params(download_queue: ::Homebrew::DownloadQueue, stale_seconds: T.nilable(Integer), enqueue: T::Boolean)
           .returns([T.any(T::Array[T.untyped], T::Hash[String, T.untyped]), T::Boolean])
       }
-      def self.fetch_api!(download_queue: nil, stale_seconds: nil)
-        Homebrew::API.fetch_json_api_file DEFAULT_API_FILENAME, stale_seconds:, download_queue:
+      def self.fetch_api!(download_queue: Homebrew.default_download_queue, stale_seconds: nil, enqueue: false)
+        Homebrew::API.fetch_json_api_file DEFAULT_API_FILENAME, stale_seconds:, download_queue:, enqueue:
       end
 
       sig {
-        params(download_queue: T.nilable(::Homebrew::DownloadQueue), stale_seconds: T.nilable(Integer))
+        params(download_queue: ::Homebrew::DownloadQueue, stale_seconds: T.nilable(Integer), enqueue: T::Boolean)
           .returns([T.any(T::Array[T.untyped], T::Hash[String, T.untyped]), T::Boolean])
       }
-      def self.fetch_tap_migrations!(download_queue: nil, stale_seconds: nil)
-        Homebrew::API.fetch_json_api_file "cask_tap_migrations.jws.json", stale_seconds:, download_queue:
+      def self.fetch_tap_migrations!(download_queue: Homebrew.default_download_queue, stale_seconds: nil,
+                                     enqueue: false)
+        Homebrew::API.fetch_json_api_file "cask_tap_migrations.jws.json", stale_seconds:, download_queue:, enqueue:
       end
 
       sig { returns(T::Boolean) }

Library/Homebrew/api/formula.rb

@@ -25,20 +25,25 @@ def self.formula_json(name)
         cache.fetch("formula_json").fetch(name)
       end
 
-      sig { params(name: String, download_queue: T.nilable(DownloadQueue)).void }
-      def self.fetch_formula_json!(name, download_queue: nil)
+      sig { params(name: String).void }
+      def self.fetch_formula_json!(name)
         endpoint = "formula/#{name}.json"
-        json_formula, updated = Homebrew::API.fetch_json_api_file endpoint, download_queue: download_queue
-        return if download_queue
+        json_formula, updated = Homebrew::API.fetch_json_api_file endpoint
 
         json_formula = JSON.parse((HOMEBREW_CACHE_API/endpoint).read) unless updated
 
         cache["formula_json"] ||= {}
         cache["formula_json"][name] = json_formula
       end
 
-      sig { params(formula: ::Formula, download_queue: T.nilable(Homebrew::DownloadQueue)).returns(Homebrew::API::SourceDownload) }
-      def self.source_download(formula, download_queue: nil)
+      sig {
+        params(
+          formula:        ::Formula,
+          download_queue: Homebrew::DownloadQueue,
+          enqueue:        T::Boolean,
+        ).returns(Homebrew::API::SourceDownload)
+      }
+      def self.source_download(formula, download_queue: Homebrew.default_download_queue, enqueue: false)
         path = formula.ruby_source_path || "Formula/#{formula.name}.rb"
         git_head = formula.tap_git_head || "HEAD"
         tap = formula.tap&.full_name || "Homebrew/homebrew-core"
@@ -49,7 +54,7 @@ def self.source_download(formula, download_queue: nil)
           cache: HOMEBREW_CACHE_API_SOURCE/"#{tap}/#{git_head}/Formula",
         )
 
-        if download_queue
+        if enqueue
           download_queue.enqueue(download)
         elsif !download.symlink_location.exist?
           download.fetch
@@ -76,19 +81,20 @@ def self.cached_json_file_path
       end
 
       sig {
-        params(download_queue: T.nilable(Homebrew::DownloadQueue), stale_seconds: T.nilable(Integer))
+        params(download_queue: Homebrew::DownloadQueue, stale_seconds: T.nilable(Integer), enqueue: T::Boolean)
           .returns([T.any(T::Array[T.untyped], T::Hash[String, T.untyped]), T::Boolean])
       }
-      def self.fetch_api!(download_queue: nil, stale_seconds: nil)
-        Homebrew::API.fetch_json_api_file DEFAULT_API_FILENAME, stale_seconds:, download_queue:
+      def self.fetch_api!(download_queue: Homebrew.default_download_queue, stale_seconds: nil, enqueue: false)
+        Homebrew::API.fetch_json_api_file DEFAULT_API_FILENAME, stale_seconds:, download_queue:, enqueue:
       end
 
       sig {
-        params(download_queue: T.nilable(Homebrew::DownloadQueue), stale_seconds: T.nilable(Integer))
+        params(download_queue: Homebrew::DownloadQueue, stale_seconds: T.nilable(Integer), enqueue: T::Boolean)
           .returns([T.any(T::Array[T.untyped], T::Hash[String, T.untyped]), T::Boolean])
       }
-      def self.fetch_tap_migrations!(download_queue: nil, stale_seconds: nil)
-        Homebrew::API.fetch_json_api_file "formula_tap_migrations.jws.json", stale_seconds:, download_queue:
+      def self.fetch_tap_migrations!(download_queue: Homebrew.default_download_queue, stale_seconds: nil,
+                                     enqueue: false)
+        Homebrew::API.fetch_json_api_file "formula_tap_migrations.jws.json", stale_seconds:, download_queue:, enqueue:
       end
 
       sig { returns(T::Boolean) }

Library/Homebrew/api/internal.rb

@@ -50,20 +50,24 @@ def self.cached_cask_json_file_path
       end
 
       sig {
-        params(download_queue: T.nilable(Homebrew::DownloadQueue), stale_seconds: T.nilable(Integer))
+        params(download_queue: Homebrew::DownloadQueue, stale_seconds: T.nilable(Integer), enqueue: T::Boolean)
           .returns([T::Hash[String, T.untyped], T::Boolean])
       }
-      def self.fetch_formula_api!(download_queue: nil, stale_seconds: nil)
-        json_contents, updated = Homebrew::API.fetch_json_api_file(formula_endpoint, stale_seconds:, download_queue:)
+      def self.fetch_formula_api!(download_queue: Homebrew.default_download_queue, stale_seconds: nil,
+                                  enqueue: false)
+        json_contents, updated = Homebrew::API.fetch_json_api_file(formula_endpoint, stale_seconds:, download_queue:,
+                                                                   enqueue:)
         [T.cast(json_contents, T::Hash[String, T.untyped]), updated]
       end
 
       sig {
-        params(download_queue: T.nilable(Homebrew::DownloadQueue), stale_seconds: T.nilable(Integer))
+        params(download_queue: Homebrew::DownloadQueue, stale_seconds: T.nilable(Integer), enqueue: T::Boolean)
           .returns([T::Hash[String, T.untyped], T::Boolean])
       }
-      def self.fetch_cask_api!(download_queue: nil, stale_seconds: nil)
-        json_contents, updated = Homebrew::API.fetch_json_api_file(cask_endpoint, stale_seconds:, download_queue:)
+      def self.fetch_cask_api!(download_queue: Homebrew.default_download_queue, stale_seconds: nil,
+                               enqueue: false)
+        json_contents, updated = Homebrew::API.fetch_json_api_file(cask_endpoint, stale_seconds:, download_queue:,
+                                                                   enqueue:)
         [T.cast(json_contents, T::Hash[String, T.untyped]), updated]
       end
 

Library/Homebrew/brew.rb

@@ -90,7 +90,7 @@
     cmd_class = Homebrew::AbstractCommand.command(cmd)
     Homebrew.running_command = cmd
     if cmd_class
-      if !Homebrew::EnvConfig.no_install_from_api? && Homebrew::EnvConfig.download_concurrency > 1
+      unless Homebrew::EnvConfig.no_install_from_api?
         require "api"
         Homebrew::API.fetch_api_files!
       end

Library/Homebrew/cask/installer.rb

@@ -25,14 +25,15 @@ class Installer
         skip_cask_deps: T::Boolean, binaries: T::Boolean, verbose: T::Boolean, zap: T::Boolean,
         require_sha: T::Boolean, upgrade: T::Boolean, reinstall: T::Boolean, installed_as_dependency: T::Boolean,
         installed_on_request: T::Boolean, quarantine: T::Boolean, verify_download_integrity: T::Boolean,
-        quiet: T::Boolean, download_queue: T.nilable(Homebrew::DownloadQueue)
+        quiet: T::Boolean, download_queue: Homebrew::DownloadQueue, defer_fetch: T::Boolean
       ).void
     }
     def initialize(cask, command: SystemCommand, force: false, adopt: false,
                    skip_cask_deps: false, binaries: true, verbose: false,
                    zap: false, require_sha: false, upgrade: false, reinstall: false,
                    installed_as_dependency: false, installed_on_request: true,
-                   quarantine: true, verify_download_integrity: true, quiet: false, download_queue: nil)
+                   quarantine: true, verify_download_integrity: true, quiet: false,
+                   download_queue: Homebrew.default_download_queue, defer_fetch: false)
       @cask = cask
       @command = command
       @force = force
@@ -50,6 +51,7 @@ def initialize(cask, command: SystemCommand, force: false, adopt: false,
       @verify_download_integrity = verify_download_integrity
       @quiet = quiet
       @download_queue = download_queue
+      @defer_fetch = T.let(defer_fetch, T::Boolean)
       @ran_prelude = T.let(false, T::Boolean)
     end
 
@@ -118,7 +120,7 @@ def fetch(quiet: nil, timeout: nil)
       forbidden_cask_and_formula_check
       forbidden_cask_artifacts_check
 
-      download(quiet:, timeout:) if @download_queue.nil?
+      download(quiet:, timeout:) unless @defer_fetch
 
       satisfy_cask_and_formula_dependencies
     end
@@ -853,15 +855,14 @@ def prelude
     sig { void }
     def enqueue_downloads
       download_queue = @download_queue
-      return if download_queue.nil?
 
       # FIXME: We need to load Cask source before enqueuing to support
       # language-specific URLs, but this will block the main process.
       if cask_from_source_api?
         if @cask.languages.any?
           load_cask_from_source_api!
         else
-          Homebrew::API::Cask.source_download(@cask, download_queue:)
+          Homebrew::API::Cask.source_download(@cask, download_queue:, enqueue: true)
         end
       end
 

Library/Homebrew/cask/reinstall.rb

@@ -10,7 +10,8 @@ class Reinstall
     sig {
       params(
         casks: ::Cask::Cask, verbose: T::Boolean, force: T::Boolean, skip_cask_deps: T::Boolean, binaries: T::Boolean,
-        require_sha: T::Boolean, quarantine: T::Boolean, zap: T::Boolean
+        require_sha: T::Boolean, quarantine: T::Boolean, zap: T::Boolean, skip_prefetch: T::Boolean,
+        download_queue: T.nilable(Homebrew::DownloadQueue)
       ).void
     }
     def self.reinstall_casks(
@@ -21,25 +22,51 @@ def self.reinstall_casks(
       binaries: false,
       require_sha: false,
       quarantine: false,
-      zap: false
+      zap: false,
+      skip_prefetch: false,
+      download_queue: nil
     )
       require "cask/installer"
 
       quarantine = true if quarantine.nil?
-
-      download_queue = Homebrew::DownloadQueue.new_if_concurrency_enabled(pour: true)
-      cask_installers = casks.map do |cask|
-        Installer.new(cask, binaries:, verbose:, force:, skip_cask_deps:, require_sha:, reinstall: true,
-                      quarantine:, zap:, download_queue:)
+      created_download_queue = T.let(false, T::Boolean)
+      if download_queue.nil?
+        if skip_prefetch
+          download_queue = Homebrew.default_download_queue
+        else
+          download_queue = Homebrew::DownloadQueue.new(pour: true)
+          created_download_queue = true
+        end
       end
 
-      if download_queue
-        cask_installers.each(&:prelude)
+      cask_installers = T.let([], T::Array[Installer])
+      begin
+        cask_installers = casks.map do |cask|
+          Installer.new(
+            cask,
+            binaries:,
+            verbose:,
+            force:,
+            skip_cask_deps:,
+            require_sha:,
+            reinstall:      true,
+            quarantine:,
+            zap:,
+            download_queue:,
+            defer_fetch:    true,
+          )
+        end
+
+        unless skip_prefetch
+          cask_installers.each(&:prelude)
 
-        oh1 "Fetching downloads for: #{casks.map { |cask| Formatter.identifier(cask.full_name) }.to_sentence}",
-            truncate: false
-        cask_installers.each(&:enqueue_downloads)
-        download_queue.fetch
+          oh1 "Fetching downloads for: #{casks.map { |cask| Formatter.identifier(cask.full_name) }.to_sentence}",
+              truncate: false
+          cask_installers.each(&:enqueue_downloads)
+          download_queue.fetch
+        end
+      ensure
+        download_queue.shutdown if created_download_queue
       end
 
       exit 1 if Homebrew.failed?