Auto merge of #4561 - domcleal:4519-concurrent-ci-updater, r=indirect

Safely store concurrent compact index downloads

When bundler is run concurrently using the same bundle dir in $HOME,
the versions file can be updated from two processes at once. The
download has been changed to a temporary file, which is securely moved
into place over the original.

If retrying the update operation, the original file is no longer
immediately deleted and instead a full download is performed, later
overwriting the original file if successful.

If two processes are updating in parallel, this should ensure the
original file isn't corrupted and that both processes succeed.

- Fixes #4519

---

This would be useful on 1.12.x if possible, since the new caching behaviour with a shared home directory is causing the errors described in #4519.

lib/bundler/vendor/compact_index_client/lib/compact_index_client/updater.rb

@@ -1,5 +1,7 @@
 # frozen_string_literal: true
+require "fileutils"
 require "stringio"
+require "tmpdir"
 require "zlib"
 
 class Bundler::CompactIndexClient
@@ -24,33 +26,41 @@ def initialize(fetcher)
     def update(local_path, remote_path, retrying = nil)
       headers = {}
 
-      if local_path.file?
-        headers["If-None-Match"] = etag_for(local_path)
-        headers["Range"] = "bytes=#{local_path.size}-"
-      else
-        # Fastly ignores Range when Accept-Encoding: gzip is set
-        headers["Accept-Encoding"] = "gzip"
-      end
+      Dir.mktmpdir(local_path.basename.to_s, local_path.dirname) do |local_temp_dir|
+        local_temp_path = Pathname.new(local_temp_dir).join(local_path.basename)
 
-      response = @fetcher.call(remote_path, headers)
-      return if response.is_a?(Net::HTTPNotModified)
+        # download new file if retrying
+        if retrying.nil? && local_path.file?
+          FileUtils.cp local_path, local_temp_path
+          headers["If-None-Match"] = etag_for(local_temp_path)
+          headers["Range"] = "bytes=#{local_temp_path.size}-"
+        else
+          # Fastly ignores Range when Accept-Encoding: gzip is set
+          headers["Accept-Encoding"] = "gzip"
+        end
 
-      content = response.body
-      if response["Content-Encoding"] == "gzip"
-        content = Zlib::GzipReader.new(StringIO.new(content)).read
-      end
+        response = @fetcher.call(remote_path, headers)
+        return if response.is_a?(Net::HTTPNotModified)
+
+        content = response.body
+        if response["Content-Encoding"] == "gzip"
+          content = Zlib::GzipReader.new(StringIO.new(content)).read
+        end
 
-      mode = response.is_a?(Net::HTTPPartialContent) ? "a" : "w"
-      local_path.open(mode) {|f| f << content }
+        mode = response.is_a?(Net::HTTPPartialContent) ? "a" : "w"
+        local_temp_path.open(mode) {|f| f << content }
 
-      response_etag = response["ETag"]
-      return if etag_for(local_path) == response_etag
+        response_etag = response["ETag"]
+        if etag_for(local_temp_path) == response_etag
+          FileUtils.mv(local_temp_path, local_path)
+          return
+        end
 
-      if retrying.nil?
-        local_path.delete
-        update(local_path, remote_path, :retrying)
-      else
-        raise MisMatchedChecksumError.new(remote_path, response_etag, etag_for(local_path))
+        if retrying.nil?
+          update(local_path, remote_path, :retrying)
+        else
+          raise MisMatchedChecksumError.new(remote_path, response_etag, etag_for(local_temp_path))
+        end
       end
     end
 

spec/install/gems/compact_index_spec.rb

@@ -656,4 +656,22 @@ def start
       bundle! :install, :artifice => "compact_index_forbidden"
     end
   end
+
+  it "performs partial update while local cache is updated by another process" do
+    gemfile <<-G
+      source "#{source_uri}"
+      gem 'rack'
+    G
+
+    # Create an empty file to trigger a partial download
+    versions = File.join(Bundler.rubygems.user_home, ".bundle", "cache", "compact_index",
+      "localgemserver.test.80.dd34752a738ee965a2a4298dc16db6c5", "versions")
+    FileUtils.mkdir_p(File.dirname(versions))
+    FileUtils.touch(versions)
+
+    bundle! :install, :artifice => "compact_index_concurrent_download"
+
+    expect(File.read(versions)).to start_with("created_at")
+    should_be_installed "rack 1.0.0"
+  end
 end

spec/support/artifice/compact_index.rb

@@ -40,7 +40,7 @@ def requested_range_for(response_body)
 
       if ranges
         status 206
-        body ranges.map! {|range| response_body.byteslice(range) }.join
+        body ranges.map! {|range| slice_body(response_body, range) }.join
       else
         status 200
         body response_body
@@ -55,6 +55,14 @@ def parse_etags(value)
       value ? value.split(/, ?/).select {|s| s.sub!(/"(.*)"/, '\1') } : []
     end
 
+    def slice_body(body, range)
+      if body.respond_to?(:byteslice)
+        body.byteslice(range)
+      else # pre-1.9.3
+        body.unpack("@#{range.first}a#{range.end + 1}").first
+      end
+    end
+
     def gems(gem_repo = gem_repo1)
       @gems ||= {}
       @gems[gem_repo] ||= begin

spec/support/artifice/compact_index_concurrent_download.rb

@@ -0,0 +1,31 @@
+# frozen_string_literal: true
+require File.expand_path("../compact_index", __FILE__)
+
+Artifice.deactivate
+
+class CompactIndexConcurrentDownload < CompactIndexAPI
+  get "/versions" do
+    versions = File.join(Bundler.rubygems.user_home, ".bundle", "cache", "compact_index",
+      "localgemserver.test.80.dd34752a738ee965a2a4298dc16db6c5", "versions")
+
+    # Verify the original (empty) content hasn't been deleted, e.g. on a retry
+    File.read(versions) == "" || raise("Original file should be present and empty")
+
+    # Verify this is only requested once for a partial download
+    env["HTTP_RANGE"] || raise("Missing Range header for expected partial download")
+
+    # Overwrite the file in parallel, which should be then overwritten
+    # after a successful download to prevent corruption
+    File.open(versions, "w") {|f| f.puts "another process" }
+
+    etag_response do
+      file = tmp("versions.list")
+      file.delete if file.file?
+      file = CompactIndex::VersionsFile.new(file.to_s)
+      file.update_with(gems)
+      CompactIndex.versions(file, nil, {})
+    end
+  end
+end
+
+Artifice.activate_with(CompactIndexConcurrentDownload)