Add distributed lock around design management mutations

This adds a distributed lock managed in Redis around all new
design management version creations, in order to prevent concurrent
commits to master in the design repository for each project.

The lock is acquired scoped to the project ID, meaning different
projects can take different locks concurrently, but design
uploads/deletions in the same repository block each other. Each mutation
is given 5 seconds of exclusive lock time to run, and an exponential
back-off is used to handle backpressure.

ExclusiveLeaseHelpers is modified to support exponential back-off

ee/app/models/design_management/version.rb

@@ -5,6 +5,7 @@ module DesignManagement
     include Importable
     include ShaAttribute
     include Gitlab::Utils::StrongMemoize
+    extend Gitlab::ExclusiveLeaseHelpers
 
     NotSameIssue = Class.new(StandardError)
 
@@ -93,6 +94,17 @@ module DesignManagement
       raise CouldNotCreateVersion.new(sha, issue_id, design_actions)
     end
 
+    CREATION_TTL = 5.seconds
+    RETRY_DELAY = ->(num) { 0.2.seconds * num**2 }
+
+    def self.lock_for_creation(project_id, &block)
+      key = "lock_for_creation:#{name}:{#{project_id}}"
+
+      in_lock(key, ttl: CREATION_TTL, retries: 5, sleep_sec: RETRY_DELAY) do |_retried|
+        yield
+      end
+    end
+
     def designs_by_event
       actions
         .includes(:design)

ee/app/services/design_management/delete_designs_service.rb

@@ -14,8 +14,11 @@ module DesignManagement
     def execute
       return error('Forbidden!') unless can_delete_designs?
 
-      actions = build_actions
-      version = run_actions(actions)
+      version = delete_designs!
+
+      version.run_after_commit do
+        ::DesignManagement::NewVersionWorker.perform_async(version.id)
+      end
 
       # Create a Geo event so changes will be replicated to secondary node(s)
       repository.log_geo_updated_event
@@ -37,6 +40,12 @@ module DesignManagement
 
     attr_reader :designs
 
+    def delete_designs!
+      DesignManagement::Version.lock_for_creation(project.id) do
+        run_actions(build_actions)
+      end
+    end
+
     def can_delete_designs?
       Ability.allowed?(current_user, :destroy_design, issue)
     end

ee/app/services/design_management/save_designs_service.rb

@@ -35,12 +35,24 @@ module DesignManagement
     attr_reader :files
 
     def upload_designs!
-      actions = build_actions
-      return [] if actions.empty?
-
-      version = run_actions(actions)
-      ::DesignManagement::NewVersionWorker.perform_async(version.id)
+      # puts "Waiting [#{Thread.current.object_id}]"
+      actions = ::DesignManagement::Version.lock_for_creation(project.id) do
+        # puts ". Building [#{Thread.current.object_id}]"
+        actions = build_actions
+        # if actions.empty?
+        #   puts ".. Skipping [#{Thread.current.object_id}]"
+        # else
+        if actions.present?
+          # puts ".. Running [#{Thread.current.object_id}]"
+          version = run_actions(actions)
+          version.run_after_commit do
+            ::DesignManagement::NewVersionWorker.perform_async(version.id)
+          end
+        end
 
+        actions
+      end
+      # puts "Done [#{Thread.current.object_id}]"
       actions.map(&:design)
     end
 

ee/spec/graphql/mutations/design_management/upload_spec.rb

@@ -12,9 +12,39 @@ describe Mutations::DesignManagement::Upload do
     described_class.new(object: nil, context: { current_user: user }, field: nil)
   end
 
-  def run_mutation(fs = files)
+  def unique_file(upload)
+    ::Gitlab::FileUpload.new(upload).tap { |f| f.original_filename = generate(:jpeg_file) }
+  end
+
+  def run_mutation(files_to_upload = files, project_path = project.full_path, iid = issue.iid)
     mutation = described_class.new(object: nil, context: { current_user: user }, field: nil)
-    mutation.resolve(project_path: project.full_path, iid: issue.iid, files: fs)
+    mutation.resolve(project_path: project_path, iid: iid, files: files_to_upload)
+  end
+
+  def parallel(blocks)
+    thread_pool = Concurrent::FixedThreadPool.new(
+      [2, Concurrent.processor_count - 1].max, { max_queue: blocks.size }
+    )
+    opts = { executor: thread_pool }
+
+    error = Concurrent::MVar.new
+
+    blocks.map { |block| Concurrent::Future.execute(opts, &block) }.each do |future|
+      future.wait(20)
+
+      if future.complete?
+        error.put(future.reason) if future.reason && error.empty?
+      else
+        future.cancel
+        error.put(StandardError.new(:cancelled)) if error.empty?
+      end
+    end
+
+    raise error.take if error.full?
+  ensure
+    thread_pool.shutdown
+    thread_pool.wait_for_termination(10)
+    thread_pool.kill if thread_pool.running?
   end
 
   describe "#resolve" do
@@ -40,32 +70,54 @@ describe Mutations::DesignManagement::Upload do
       end
 
       describe 'contention in the design repo' do
+        before do
+          issue.design_collection.repository.create_if_not_exists
+        end
+
         let(:files) do
-          [
-            fixture_file_upload('spec/fixtures/dk.png'),
-            fixture_file_upload('spec/fixtures/rails_sample.jpg'),
-            fixture_file_upload('spec/fixtures/banana_sample.gif')
-          ].cycle(20).to_a
+          ['dk.png', 'rails_sample.jpg', 'banana_sample.gif']
+           .cycle
+           .take(Concurrent.processor_count * 2)
+           .map { |f| unique_file(fixture_file_upload("spec/fixtures/#{f}")) }
+        end
+
+        def creates_designs
+          prior_count = DesignManagement::Design.count
+
+          expect { yield }.not_to raise_error
+
+          expect(DesignManagement::Design.count).to eq(prior_count + files.size)
         end
 
         describe 'running requests in parallel' do
           it 'does not cause errors' do
-            expect do
-              threads = files.map do |f|
-                Thread.new { run_mutation([f]) }
+            creates_designs do
+              parallel(files.map { |f| -> { run_mutation([f]) } })
+            end
+          end
+        end
+
+        describe 'running requests in parallel on different issues' do
+          it 'does not cause errors' do
+            creates_designs do
+              issues = create_list(:issue, files.size, author: user)
+              issues.each { |i| i.project.add_developer(user) }
+              blocks = files.zip(issues).map do |(f, i)|
+                -> { run_mutation([f], i.project.full_path, i.iid) }
               end
-              threads.each(&:join)
-            end.not_to raise_error
+
+              parallel(blocks)
+            end
           end
         end
 
         describe 'running requests in serial' do
           it 'does not cause errors' do
-            expect do
+            creates_designs do
               files.each do |f|
                 run_mutation([f])
               end
-            end.not_to raise_error
+            end
           end
         end
       end

lib/gitlab/exclusive_lease_helpers.rb

@@ -16,11 +16,15 @@ module Gitlab
 
       lease = Gitlab::ExclusiveLease.new(key, timeout: ttl)
       retried = false
+      max_attempts = 1 + retries
 
       until uuid = lease.try_obtain
         # Keep trying until we obtain the lease. To prevent hammering Redis too
         # much we'll wait for a bit.
-        sleep(sleep_sec)
+        attempt_number = max_attempts - retries
+        delay = sleep_sec.respond_to?(:call) ? sleep_sec.call(attempt_number) : sleep_sec
+
+        sleep(delay)
         (retries -= 1) < 0 ? break : retried ||= true
       end
 

spec/lib/gitlab/exclusive_lease_helpers_spec.rb

@@ -82,10 +82,22 @@ describe Gitlab::ExclusiveLeaseHelpers, :clean_gitlab_redis_shared_state do
       end
 
       context 'when sleep second is specified' do
-        let(:options) { { retries: 0, sleep_sec: 0.05.seconds } }
+        let(:options) { { retries: 1, sleep_sec: 0.05.seconds } }
 
         it 'receives the specified argument' do
-          expect(class_instance).to receive(:sleep).with(0.05.seconds).once
+          expect(class_instance).to receive(:sleep).with(0.05.seconds).twice
+
+          expect { subject }.to raise_error('Failed to obtain a lock')
+        end
+      end
+
+      context 'when sleep second is specified as a lambda' do
+        let(:options) { { retries: 2, sleep_sec: ->(num) { 0.1 + num } } }
+
+        it 'receives the specified argument' do
+          expect(class_instance).to receive(:sleep).with(1.1.seconds).once
+          expect(class_instance).to receive(:sleep).with(2.1.seconds).once
+          expect(class_instance).to receive(:sleep).with(3.1.seconds).once
 
           expect { subject }.to raise_error('Failed to obtain a lock')
         end

spec/support/file_upload.rb

+# frozen_string_literal: true
+
+class Gitlab::FileUpload < SimpleDelegator
+  attr_accessor :original_filename
+end