Move write operations to a separate worker

See merge request gitlab-org/gitlab!62761

app/services/web_hook_service.rb

@@ -29,15 +29,17 @@ class WebHookService
   GITLAB_EVENT_HEADER = 'X-Gitlab-Event'
 
   attr_accessor :hook, :data, :hook_name, :request_options
+  attr_reader :uniqueness_token
 
   def self.hook_to_event(hook_name)
     hook_name.to_s.singularize.titleize
   end
 
-  def initialize(hook, data, hook_name)
+  def initialize(hook, data, hook_name, uniqueness_token = nil)
     @hook = hook
     @data = data
     @hook_name = hook_name.to_s
+    @uniqueness_token = uniqueness_token
     @request_options = {
       timeout: Gitlab.config.gitlab.webhook_timeout,
       allow_local_requests: hook.allow_local_requests?
@@ -123,10 +125,8 @@ class WebHookService
   end
 
   def log_execution(trigger:, url:, request_data:, response:, execution_duration:, error_message: nil)
-    handle_failure(response, hook)
-
-    WebHookLog.create(
-      web_hook: hook,
+    category = response_category(response)
+    log_data = {
       trigger: trigger,
       url: url,
       execution_duration: execution_duration,
@@ -136,16 +136,19 @@ class WebHookService
       response_body: safe_response_body(response),
       response_status: response.code,
       internal_error_message: error_message
-    )
+    }
+
+    ::WebHooks::LogExecutionWorker
+      .perform_async(hook.id, log_data, category, uniqueness_token)
   end
 
-  def handle_failure(response, hook)
+  def response_category(response)
     if response.success? || response.redirection?
-      hook.enable!
+      :ok
     elsif response.internal_server_error?
-      hook.backoff!
+      :error
     else
-      hook.failed!
+      :failed
     end
   end
 

app/services/web_hooks/log_execution_service.rb

+# frozen_string_literal: true
+
+module WebHooks
+  class LogExecutionService
+    attr_reader :hook, :log_data, :response_category
+
+    def initialize(hook:, log_data:, response_category:)
+      @hook = hook
+      @log_data = log_data
+      @response_category = response_category
+    end
+
+    def execute
+      update_hook_executability
+      log_execution
+    end
+
+    private
+
+    def log_execution
+      WebHookLog.create!(web_hook: hook, **log_data.transform_keys(&:to_sym))
+    end
+
+    def update_hook_executability
+      case response_category
+      when :ok
+        hook.enable!
+      when :error
+        hook.backoff!
+      when :failed
+        hook.failed!
+      end
+    end
+  end
+end

app/workers/all_queues.yml

@@ -2828,6 +2828,15 @@
   :idempotent: true
   :tags:
   - :exclude_from_kubernetes
+- :name: web_hooks_log_execution
+  :worker_name: WebHooks::LogExecutionWorker
+  :feature_category: :integrations
+  :has_external_dependencies:
+  :urgency: :low
+  :resource_boundary: :unknown
+  :weight: 1
+  :idempotent: true
+  :tags: []
 - :name: wikis_git_garbage_collect
   :worker_name: Wikis::GitGarbageCollectWorker
   :feature_category: :gitaly

app/workers/web_hook_worker.rb

@@ -16,7 +16,7 @@ class WebHookWorker
     hook = WebHook.find(hook_id)
     data = data.with_indifferent_access
 
-    WebHookService.new(hook, data, hook_name).execute
+    WebHookService.new(hook, data, hook_name, jid).execute
   end
 end
 # rubocop:enable Scalability/IdempotentWorker

app/workers/web_hooks/log_execution_worker.rb

+# frozen_string_literal: true
+
+module WebHooks
+  class LogExecutionWorker
+    include ApplicationWorker
+
+    idempotent!
+    feature_category :integrations
+    urgency :low
+
+    # This worker accepts an extra argument. This enables us to
+    # treat this worker as idempotent. Currently this is set to
+    # the Job ID (jid) of the parent worker.
+    def perform(hook_id, log_data, response_category, _unique_by)
+      hook = WebHook.find_by_id(hook_id)
+
+      return unless hook # hook has been deleted before we could run.
+
+      ::WebHooks::LogExecutionService
+        .new(hook: hook, log_data: log_data, response_category: response_category.to_sym)
+        .execute
+    end
+  end
+end

config/sidekiq_queues.yml

@@ -400,6 +400,8 @@
   - 1
 - - web_hooks_destroy
   - 1
+- - web_hooks_log_execution
+  - 1
 - - wikis_git_garbage_collect
   - 1
 - - x509_certificate_revoke

spec/services/web_hook_service_spec.rb

@@ -174,13 +174,19 @@ RSpec.describe WebHookService do
     context 'execution logging' do
       let(:hook_log) { project_hook.web_hook_logs.last }
 
+      def run_service
+        service_instance.execute
+        ::WebHooks::LogExecutionWorker.drain
+        project_hook.reload
+      end
+
       context 'with success' do
         before do
           stub_full_request(project_hook.url, method: :post).to_return(status: 200, body: 'Success')
         end
 
         it 'log successful execution' do
-          service_instance.execute
+          run_service
 
           expect(hook_log.trigger).to eq('push_hooks')
           expect(hook_log.url).to eq(project_hook.url)
@@ -191,12 +197,16 @@ RSpec.describe WebHookService do
           expect(hook_log.internal_error_message).to be_nil
         end
 
+        it 'does not log in the service itself' do
+          expect { service_instance.execute }.not_to change(::WebHookLog, :count)
+        end
+
         it 'does not increment the failure count' do
-          expect { service_instance.execute }.not_to change(project_hook, :recent_failures)
+          expect { run_service }.not_to change(project_hook, :recent_failures)
         end
 
         it 'does not change the disabled_until attribute' do
-          expect { service_instance.execute }.not_to change(project_hook, :disabled_until)
+          expect { run_service }.not_to change(project_hook, :disabled_until)
         end
 
         context 'when the hook had previously failed' do
@@ -205,7 +215,7 @@ RSpec.describe WebHookService do
           end
 
           it 'resets the failure count' do
-            expect { service_instance.execute }.to change(project_hook, :recent_failures).to(0)
+            expect { run_service }.to change(project_hook, :recent_failures).to(0)
           end
         end
       end
@@ -216,7 +226,7 @@ RSpec.describe WebHookService do
         end
 
         it 'logs failed execution' do
-          service_instance.execute
+          run_service
 
           expect(hook_log).to have_attributes(
             trigger: eq('push_hooks'),
@@ -230,17 +240,17 @@ RSpec.describe WebHookService do
         end
 
         it 'increments the failure count' do
-          expect { service_instance.execute }.to change(project_hook, :recent_failures).by(1)
+          expect { run_service }.to change(project_hook, :recent_failures).by(1)
         end
 
         it 'does not change the disabled_until attribute' do
-          expect { service_instance.execute }.not_to change(project_hook, :disabled_until)
+          expect { run_service }.not_to change(project_hook, :disabled_until)
         end
 
         it 'does not allow the failure count to overflow' do
           project_hook.update!(recent_failures: 32767)
 
-          expect { service_instance.execute }.not_to change(project_hook, :recent_failures)
+          expect { run_service }.not_to change(project_hook, :recent_failures)
         end
 
         context 'when the web_hooks_disable_failed FF is disabled' do
@@ -252,7 +262,7 @@ RSpec.describe WebHookService do
           it 'does not allow the failure count to overflow' do
             project_hook.update!(recent_failures: 32767)
 
-            expect { service_instance.execute }.not_to change(project_hook, :recent_failures)
+            expect { run_service }.not_to change(project_hook, :recent_failures)
           end
         end
       end
@@ -263,7 +273,7 @@ RSpec.describe WebHookService do
         end
 
         it 'log failed execution' do
-          service_instance.execute
+          run_service
 
           expect(hook_log.trigger).to eq('push_hooks')
           expect(hook_log.url).to eq(project_hook.url)
@@ -275,17 +285,15 @@ RSpec.describe WebHookService do
         end
 
         it 'does not increment the failure count' do
-          expect { service_instance.execute }.not_to change(project_hook, :recent_failures)
+          expect { run_service }.not_to change(project_hook, :recent_failures)
         end
 
         it 'backs off' do
-          expect(project_hook).to receive(:backoff!).and_call_original
-
-          expect { service_instance.execute }.to change(project_hook, :disabled_until)
+          expect { run_service }.to change(project_hook, :disabled_until)
         end
 
         it 'increases the backoff count' do
-          expect { service_instance.execute }.to change(project_hook, :backoff_count).by(1)
+          expect { run_service }.to change(project_hook, :backoff_count).by(1)
         end
 
         context 'when the previous cool-off was near the maximum' do
@@ -294,7 +302,7 @@ RSpec.describe WebHookService do
           end
 
           it 'sets the disabled_until attribute' do
-            expect { service_instance.execute }.to change(project_hook, :disabled_until).to(1.day.from_now)
+            expect { run_service }.to change(project_hook, :disabled_until).to(1.day.from_now)
           end
         end
 
@@ -304,7 +312,7 @@ RSpec.describe WebHookService do
           end
 
           it 'sets the disabled_until attribute' do
-            expect { service_instance.execute }.to change(project_hook, :disabled_until).to(1.day.from_now)
+            expect { run_service }.to change(project_hook, :disabled_until).to(1.day.from_now)
           end
         end
       end
@@ -312,7 +320,7 @@ RSpec.describe WebHookService do
       context 'with unsafe response body' do
         before do
           stub_full_request(project_hook.url, method: :post).to_return(status: 200, body: "\xBB")
-          service_instance.execute
+          run_service
         end
 
         it 'log successful execution' do

spec/workers/web_hook_worker_spec.rb

@@ -10,7 +10,7 @@ RSpec.describe WebHookWorker do
 
   describe '#perform' do
     it 'delegates to WebHookService' do
-      expect_next(WebHookService, project_hook, data.with_indifferent_access, hook_name).to receive(:execute)
+      expect_next(WebHookService, project_hook, data.with_indifferent_access, hook_name, anything).to receive(:execute)
 
       subject.perform(project_hook.id, data, hook_name)
     end