Fallback to sequentially fetching specs on 429s

If the compact index returns TooManyRequests, take it easier by
requesting dependencies sequentially instead.

4 files changed, 50 insertions, 16 deletions

diff --git a/lib/bundler/compact_index_client.rb b/lib/bundler/compact_index_client.rb
index 6c241ca07a..2f713041c8 100644
--- a/lib/bundler/compact_index_client.rb
+++ b/lib/bundler/compact_index_client.rb
@@ -18,11 +18,6 @@ module Bundler
 
     attr_reader :directory
 
-    # @return [Lambda] A lambda that takes an array of inputs and a block, and
-    #         maps the inputs with the block in parallel.
-    #
-    attr_accessor :in_parallel
-
     def initialize(directory, fetcher)
       @directory = Pathname.new(directory)
       @updater = Updater.new(fetcher)
@@ -31,7 +26,28 @@ module Bundler
       @info_checksums_by_name = {}
       @parsed_checksums = false
       @mutex = Mutex.new
-      @in_parallel = lambda do |inputs, &blk|
+    end
+
+    def execution_mode=(block)
+      Bundler::CompactIndexClient.debug { "execution_mode=" }
+      @endpoints = Set.new
+
+      @execution_mode = block
+    end
+
+    # @return [Lambda] A lambda that takes an array of inputs and a block, and
+    #         maps the inputs with the block in parallel.
+    #
+    def execution_mode
+      @execution_mode || sequentially
+    end
+
+    def sequential_execution_mode!
+      self.execution_mode = sequentially
+    end
+
+    def sequentially
+      @sequentially ||= lambda do |inputs, &blk|
         inputs.map(&blk)
       end
     end
@@ -51,7 +67,7 @@ module Bundler
 
     def dependencies(names)
       Bundler::CompactIndexClient.debug { "dependencies(#{names})" }
-      in_parallel.call(names) do |name|
+      execution_mode.call(names) do |name|
         update_info(name)
         @cache.dependencies(name).map {|d| d.unshift(name) }
       end.flatten(1)
diff --git a/lib/bundler/fetcher.rb b/lib/bundler/fetcher.rb
index 7df42adc34..272aac6882 100644
--- a/lib/bundler/fetcher.rb
+++ b/lib/bundler/fetcher.rb
@@ -15,6 +15,8 @@ module Bundler
 
     # This error is raised when it looks like the network is down
     class NetworkDownError < HTTPError; end
+    # This error is raised if we should rate limit our requests to the API
+    class TooManyRequestsError < HTTPError; end
     # This error is raised if the API returns a 413 (only printed in verbose)
     class FallbackError < HTTPError; end
     # This is the error raised if OpenSSL fails the cert verification
diff --git a/lib/bundler/fetcher/compact_index.rb b/lib/bundler/fetcher/compact_index.rb
index cfc74d642c..d462aa1463 100644
--- a/lib/bundler/fetcher/compact_index.rb
+++ b/lib/bundler/fetcher/compact_index.rb
@@ -39,7 +39,13 @@ module Bundler
         until remaining_gems.empty?
           log_specs "Looking up gems #{remaining_gems.inspect}"
 
-          deps = compact_index_client.dependencies(remaining_gems)
+          deps = begin
+            parallel_compact_index_client.dependencies(remaining_gems)
+          rescue TooManyRequestsError
+            @bundle_worker.stop if @bundle_worker
+            @bundle_worker = nil # reset it.  Not sure if necessary
+            serial_compact_index_client.dependencies(remaining_gems)
+          end
           next_gems = deps.map {|d| d[3].map(&:first).flatten(1) }.flatten(1).uniq
           deps.each {|dep| gem_info << dep }
           complete_gems.concat(deps.map(&:first)).uniq!
@@ -80,18 +86,26 @@ module Bundler
     private
 
       def compact_index_client
-        @compact_index_client ||= begin
+        @compact_index_client ||=
           SharedHelpers.filesystem_access(cache_path) do
             CompactIndexClient.new(cache_path, client_fetcher)
-          end.tap do |client|
-            client.in_parallel = lambda do |inputs, &blk|
-              func = lambda {|object, _index| blk.call(object) }
-              worker = bundle_worker(func)
-              inputs.each {|input| worker.enq(input) }
-              inputs.map { worker.deq }
-            end
           end
+      end
+
+      def parallel_compact_index_client
+        compact_index_client.execution_mode = lambda do |inputs, &blk|
+          func = lambda {|object, _index| blk.call(object) }
+          worker = bundle_worker(func)
+          inputs.each {|input| worker.enq(input) }
+          inputs.map { worker.deq }
         end
+
+        compact_index_client
+      end
+
+      def serial_compact_index_client
+        compact_index_client.sequential_execution_mode!
+        compact_index_client
       end
 
       def bundle_worker(func = nil)
diff --git a/lib/bundler/fetcher/downloader.rb b/lib/bundler/fetcher/downloader.rb
index 87ad4140fd..2aeb9962c4 100644
--- a/lib/bundler/fetcher/downloader.rb
+++ b/lib/bundler/fetcher/downloader.rb
@@ -34,6 +34,8 @@ module Bundler
           fetch(uri, new_headers)
         when Net::HTTPRequestEntityTooLarge
           raise FallbackError, response.body
+        when Net::HTTPTooManyRequests
+          raise TooManyRequestsError, response.body
         when Net::HTTPUnauthorized
           raise AuthenticationRequiredError, uri.host
         when Net::HTTPNotFound