Merge branch 'feature/gb/idempotent-build-trace-chunks-flush-worker' into 'master'

Make build trace chunks flush worker idempotent

See merge request gitlab-org/gitlab!41579

app/models/ci/build_trace_chunk.rb

@@ -108,6 +108,14 @@ module Ci
       Ci::BuildTraceChunkFlushWorker.perform_async(id)
     end
 
+    def persisted?
+      !redis?
+    end
+
+    def live?
+      redis?
+    end
+
     private
 
     def get_data
@@ -170,14 +178,6 @@ module Ci
       save! if changed?
     end
 
-    def persisted?
-      !redis?
-    end
-
-    def live?
-      redis?
-    end
-
     def full?
       size == CHUNK_SIZE
     end

app/workers/all_queues.yml

@@ -897,7 +897,7 @@
   :urgency: :low
   :resource_boundary: :unknown
   :weight: 1
-  :idempotent:
+  :idempotent: true
   :tags: []
 - :name: pipeline_background:ci_daily_build_group_report_results
   :feature_category: :continuous_integration

app/workers/ci/build_trace_chunk_flush_worker.rb

 # frozen_string_literal: true
 
 module Ci
-  class BuildTraceChunkFlushWorker # rubocop:disable Scalability/IdempotentWorker
+  class BuildTraceChunkFlushWorker
     include ApplicationWorker
     include PipelineBackgroundQueue
 
+    idempotent!
+
     # rubocop: disable CodeReuse/ActiveRecord
     def perform(chunk_id)
       ::Ci::BuildTraceChunk.find_by(id: chunk_id).try do |chunk|

spec/workers/ci/build_trace_chunk_flush_worker_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Ci::BuildTraceChunkFlushWorker do
+  let(:data) { 'x' * Ci::BuildTraceChunk::CHUNK_SIZE }
+
+  let(:chunk) do
+    create(:ci_build_trace_chunk, :redis_with_data, initial_data: data)
+  end
+
+  it 'migrates chunk to a permanent store' do
+    expect(chunk).to be_live
+
+    described_class.new.perform(chunk.id)
+
+    expect(chunk.reload).to be_persisted
+  end
+
+  describe '#perform' do
+    it_behaves_like 'an idempotent worker' do
+      let(:job_args) { [chunk.id] }
+
+      it 'migrates build trace chunk to a safe store' do
+        subject
+
+        expect(chunk.reload).to be_persisted
+      end
+    end
+  end
+end