Schedule CreateEvidenceWorker jobs in a sliding window

Closes https://gitlab.com/gitlab-org/gitlab/-/issues/282441

See merge request https://gitlab.com/gitlab-org/gitlab/-/merge_requests/47638

app/finders/releases/evidence_pipeline_finder.rb

+# frozen_string_literal: true
+
+module Releases
+  class EvidencePipelineFinder
+    include Gitlab::Utils::StrongMemoize
+
+    attr_reader :project, :params
+
+    def initialize(project, params = {})
+      @project = project
+      @params = params
+    end
+
+    def execute
+      # TODO: remove this with the release creation moved to it's own form https://gitlab.com/gitlab-org/gitlab/-/issues/214245
+      return params[:evidence_pipeline] if params[:evidence_pipeline]
+
+      sha = existing_tag&.dereferenced_target&.sha
+      sha ||= repository&.commit(ref)&.sha
+
+      return unless sha
+
+      project.ci_pipelines.for_sha(sha).last
+    end
+
+    private
+
+    def repository
+      strong_memoize(:repository) do
+        project.repository
+      end
+    end
+
+    def existing_tag
+      repository.find_tag(tag_name)
+    end
+
+    def tag_name
+      params[:tag]
+    end
+
+    def ref
+      params[:ref]
+    end
+  end
+end

app/models/release.rb

@@ -29,6 +29,8 @@ class Release < ApplicationRecord
   scope :preloaded, -> { includes(:evidences, :milestones, project: [:project_feature, :route, { namespace: :route }]) }
   scope :with_project_and_namespace, -> { includes(project: :namespace) }
   scope :recent, -> { sorted.limit(MAX_NUMBER_TO_DISPLAY) }
+  scope :without_evidence, -> { left_joins(:evidences).where(::Releases::Evidence.arel_table[:id].eq(nil)) }
+  scope :released_within_2hrs, -> { where(released_at: Time.zone.now - 1.hour..Time.zone.now + 1.hour) }
 
   # Sorting
   scope :order_created, -> { reorder('created_at ASC') }

app/services/releases/base_service.rb

@@ -11,8 +11,6 @@ module Releases
       @project, @current_user, @params = project, user, params.dup
     end
 
-    delegate :repository, to: :project
-
     def tag_name
       params[:tag]
     end
@@ -39,22 +37,18 @@ module Releases
       end
     end
 
-    def existing_tag
-      strong_memoize(:existing_tag) do
-        repository.find_tag(tag_name)
-      end
-    end
-
-    def tag_exist?
-      existing_tag.present?
-    end
-
     def repository
       strong_memoize(:repository) do
         project.repository
       end
     end
 
+    def existing_tag
+      strong_memoize(:existing_tag) do
+        repository.find_tag(tag_name)
+      end
+    end
+
     def milestones
       return [] unless param_for_milestone_titles_provided?
 

app/services/releases/create_service.rb

@@ -10,7 +10,7 @@ module Releases
       # should be found before the creation of new tag
       # because tag creation can spawn new pipeline
       # which won't have any data for evidence yet
-      evidence_pipeline = find_evidence_pipeline
+      evidence_pipeline = Releases::EvidencePipelineFinder.new(project, params).execute
 
       tag = ensure_tag
 
@@ -78,26 +78,10 @@ module Releases
       )
     end
 
-    def find_evidence_pipeline
-      # TODO: remove this with the release creation moved to it's own form https://gitlab.com/gitlab-org/gitlab/-/issues/214245
-      return params[:evidence_pipeline] if params[:evidence_pipeline]
-
-      sha = existing_tag&.dereferenced_target&.sha
-      sha ||= repository.commit(ref)&.sha
-
-      return unless sha
-
-      project.ci_pipelines.for_sha(sha).last
-    end
-
     def create_evidence!(release, pipeline)
-      return if release.historical_release?
+      return if release.historical_release? || release.upcoming_release?
 
-      if release.upcoming_release?
-        CreateEvidenceWorker.perform_at(release.released_at, release.id, pipeline&.id)
-      else
-        CreateEvidenceWorker.perform_async(release.id, pipeline&.id)
-      end
+      ::Releases::CreateEvidenceWorker.perform_async(release.id, pipeline&.id)
     end
   end
 end

app/workers/all_queues.yml

@@ -323,6 +323,22 @@
   :weight: 1
   :idempotent:
   :tags: []
+- :name: cronjob:releases_create_evidence
+  :feature_category: :release_evidence
+  :has_external_dependencies:
+  :urgency: :low
+  :resource_boundary: :unknown
+  :weight: 1
+  :idempotent:
+  :tags: []
+- :name: cronjob:releases_manage_evidence
+  :feature_category: :release_evidence
+  :has_external_dependencies:
+  :urgency: :low
+  :resource_boundary: :unknown
+  :weight: 1
+  :idempotent:
+  :tags: []
 - :name: cronjob:remove_expired_group_links
   :feature_category: :authentication_and_authorization
   :has_external_dependencies:
@@ -1377,14 +1393,6 @@
   :weight: 2
   :idempotent: true
   :tags: []
-- :name: create_evidence
-  :feature_category: :release_evidence
-  :has_external_dependencies:
-  :urgency: :low
-  :resource_boundary: :unknown
-  :weight: 2
-  :idempotent:
-  :tags: []
 - :name: create_note_diff_file
   :feature_category: :source_code_management
   :has_external_dependencies:

app/workers/create_evidence_worker.rb

-# frozen_string_literal: true
-
-class CreateEvidenceWorker # rubocop:disable Scalability/IdempotentWorker
-  include ApplicationWorker
-
-  feature_category :release_evidence
-  weight 2
-
-  # pipeline_id is optional for backward compatibility with existing jobs
-  # caller should always try to provide the pipeline and pass nil only
-  # if pipeline is absent
-  def perform(release_id, pipeline_id = nil)
-    release = Release.find_by_id(release_id)
-    return unless release
-
-    pipeline = Ci::Pipeline.find_by_id(pipeline_id)
-
-    ::Releases::CreateEvidenceService.new(release, pipeline: pipeline).execute
-  end
-end

app/workers/releases/create_evidence_worker.rb

+# frozen_string_literal: true
+
+module Releases
+  class CreateEvidenceWorker # rubocop:disable Scalability/IdempotentWorker
+    include ApplicationWorker
+    include CronjobQueue # rubocop:disable Scalability/CronWorkerContext
+
+    feature_category :release_evidence
+
+    # pipeline_id is optional for backward compatibility with existing jobs
+    # caller should always try to provide the pipeline and pass nil only
+    # if pipeline is absent
+    def perform(release_id, pipeline_id = nil)
+      release = Release.find_by_id(release_id)
+
+      return unless release
+
+      pipeline = Ci::Pipeline.find_by_id(pipeline_id)
+
+      ::Releases::CreateEvidenceService.new(release, pipeline: pipeline).execute
+    end
+  end
+end

app/workers/releases/manage_evidence_worker.rb

+# frozen_string_literal: true
+
+module Releases
+  class ManageEvidenceWorker # rubocop:disable Scalability/IdempotentWorker
+    include ApplicationWorker
+    include CronjobQueue # rubocop:disable Scalability/CronWorkerContext
+
+    feature_category :release_evidence
+
+    def perform
+      releases = Release.without_evidence.released_within_2hrs
+
+      releases.each do |release|
+        project = release.project
+        params = { tag: release.tag }
+
+        evidence_pipeline = Releases::EvidencePipelineFinder.new(project, params).execute
+
+        # perform_at released_at
+        ::Releases::CreateEvidenceWorker.perform_async(release.id, evidence_pipeline&.id)
+      end
+    end
+  end
+end

app/workers/trending_projects_worker.rb

@@ -2,10 +2,6 @@
 
 class TrendingProjectsWorker # rubocop:disable Scalability/IdempotentWorker
   include ApplicationWorker
-  # rubocop:disable Scalability/CronWorkerContext
-  # This worker does not perform work scoped to a context
-  include CronjobQueue
-  # rubocop:enable Scalability/CronWorkerContext
   include CronjobQueue # rubocop:disable Scalability/CronWorkerContext
 
   feature_category :source_code_management

changelogs/unreleased/282441-schedule-createevidenceworker-jobs-in-a-sliding-window.yml

+---
+title: Schedule CreateEvidenceWorker jobs in a sliding window
+merge_request: 47638
+author:
+type: added

config/initializers/1_settings.rb

@@ -532,6 +532,9 @@ Settings.cron_jobs['member_invitation_reminder_emails_worker']['job_class'] = 'M
 Settings.cron_jobs['schedule_merge_request_cleanup_refs_worker'] ||= Settingslogic.new({})
 Settings.cron_jobs['schedule_merge_request_cleanup_refs_worker']['cron'] ||= '* * * * *'
 Settings.cron_jobs['schedule_merge_request_cleanup_refs_worker']['job_class'] = 'ScheduleMergeRequestCleanupRefsWorker'
+Settings.cron_jobs['manage_evidence_worker'] ||= Settingslogic.new({})
+Settings.cron_jobs['manage_evidence_worker']['cron'] ||= '0 * * * *'
+Settings.cron_jobs['manage_evidence_worker']['job_class'] = 'Releases::ManageEvidenceWorker'
 
 Gitlab.ee do
   Settings.cron_jobs['active_user_count_threshold_worker'] ||= Settingslogic.new({})

config/sidekiq_queues.yml

@@ -58,8 +58,6 @@
   - 1
 - - create_commit_signature
   - 2
-- - create_evidence
-  - 2
 - - create_github_webhook
   - 2
 - - create_note_diff_file

db/migrate/20201029144524_add_index_to_releases.rb

+# frozen_string_literal: true
+
+class AddIndexToReleases < ActiveRecord::Migration[6.0]
+  include Gitlab::Database::MigrationHelpers
+
+  DOWNTIME = false
+  INDEX_NAME = 'index_releases_on_released_at'
+
+  disable_ddl_transaction!
+
+  def up
+    add_concurrent_index :releases, :released_at, name: INDEX_NAME
+  end
+
+  def down
+    remove_concurrent_index_by_name :releases, INDEX_NAME
+  end
+end

db/schema_migrations/20201029144524

+764f08e3083985bb8e206bd25fb27209702110bb4848c8bbfc6546a2777d9157
\ No newline at end of file

db/structure.sql

@@ -21789,6 +21789,8 @@ CREATE INDEX index_releases_on_author_id ON releases USING btree (author_id);
 
 CREATE INDEX index_releases_on_project_id_and_tag ON releases USING btree (project_id, tag);
 
+CREATE INDEX index_releases_on_released_at ON releases USING btree (released_at);
+
 CREATE INDEX index_remote_mirrors_on_last_successful_update_at ON remote_mirrors USING btree (last_successful_update_at);
 
 CREATE INDEX index_remote_mirrors_on_project_id ON remote_mirrors USING btree (project_id);

ee/lib/ee/api/releases.rb

@@ -18,7 +18,10 @@ module EE
             authorize_create_evidence!
 
             if release.present?
-              CreateEvidenceWorker.perform_async(release.id)
+              params = { tag: release.tag }
+              evidence_pipeline = ::Releases::EvidencePipelineFinder.new(release.project, params).execute
+              ::Releases::CreateEvidenceWorker.perform_async(release.id, evidence_pipeline)
+
               status :accepted
             else
               status :not_found

spec/finders/releases/evidence_pipeline_finder_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Releases::EvidencePipelineFinder, '#execute' do
+  let(:params) { {} }
+  let(:project) { create(:project, :repository) }
+  let(:tag_name) { project.repository.tag_names.first }
+  let(:sha) { project.repository.find_tag(tag_name).dereferenced_target.sha }
+  let!(:pipeline) { create(:ci_empty_pipeline, sha: sha, project: project) }
+
+  subject { described_class.new(project, params).execute }
+
+  context 'when the tag is passed' do
+    let(:params) { { tag: tag_name } }
+
+    it 'returns the evidence pipeline' do
+      expect(subject).to eq(pipeline)
+    end
+  end
+
+  context 'when the ref is passed' do
+    let(:params) { { ref: sha } }
+
+    it 'returns the evidence pipeline' do
+      expect(subject).to eq(pipeline)
+    end
+  end
+
+  context 'empty params' do
+    it 'returns nil' do
+      expect(subject).to be_nil
+    end
+  end
+
+  # TODO: remove this with the release creation moved to it's own form https://gitlab.com/gitlab-org/gitlab/-/issues/214245
+  context 'params[:evidence_pipeline] is present' do
+    let(:params) { { evidence_pipeline: pipeline } }
+
+    it 'returns the passed evidence pipeline' do
+      expect(subject).to eq(pipeline)
+    end
+  end
+end

spec/services/releases/create_service_spec.rb

@@ -217,7 +217,7 @@ RSpec.describe Releases::CreateService do
       let(:released_at) { 3.weeks.ago }
 
       it 'does not execute CreateEvidenceWorker' do
-        expect { subject }.not_to change(CreateEvidenceWorker.jobs, :size)
+        expect { subject }.not_to change(Releases::CreateEvidenceWorker.jobs, :size)
       end
 
       it 'does not create an Evidence object', :sidekiq_inline do
@@ -316,7 +316,7 @@ RSpec.describe Releases::CreateService do
       end
 
       it 'queues CreateEvidenceWorker' do
-        expect { subject }.to change(CreateEvidenceWorker.jobs, :size).by(1)
+        expect { subject }.to change(Releases::CreateEvidenceWorker.jobs, :size).by(1)
       end
 
       it 'creates Evidence', :sidekiq_inline do
@@ -341,18 +341,12 @@ RSpec.describe Releases::CreateService do
     context 'upcoming release' do
       let(:released_at) { 1.day.from_now }
 
-      it 'queues CreateEvidenceWorker' do
-        expect { subject }.to change(CreateEvidenceWorker.jobs, :size).by(1)
-      end
-
-      it 'queues CreateEvidenceWorker at the released_at timestamp' do
-        subject
-
-        expect(CreateEvidenceWorker.jobs.last['at'].to_i).to eq(released_at.to_i)
+      it 'does not execute CreateEvidenceWorker' do
+        expect { subject }.not_to change(Releases::CreateEvidenceWorker.jobs, :size)
       end
 
-      it 'creates Evidence', :sidekiq_inline do
-        expect { subject }.to change(Releases::Evidence, :count).by(1)
+      it 'does not create an Evidence object', :sidekiq_inline do
+        expect { subject }.not_to change(Releases::Evidence, :count)
       end
 
       it 'is not a historical release' do
@@ -366,8 +360,6 @@ RSpec.describe Releases::CreateService do
 
         expect(last_release.upcoming_release?).to be_truthy
       end
-
-      include_examples 'uses the right pipeline for evidence'
     end
   end
 end

spec/workers/create_evidence_worker_spec.rb → spec/workers/releases/create_evidence_worker_spec.rb

@@ -2,7 +2,7 @@
 
 require 'spec_helper'
 
-RSpec.describe CreateEvidenceWorker do
+RSpec.describe Releases::CreateEvidenceWorker do
   let(:project) { create(:project, :repository) }
   let(:release) { create(:release, project: project) }
   let(:pipeline) { create(:ci_empty_pipeline, sha: release.sha, project: project) }

spec/workers/releases/manage_evidence_worker_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Releases::ManageEvidenceWorker do
+  let(:project) { create(:project, :repository) }
+
+  shared_examples_for 'does not create a new Evidence record' do
+    specify :sidekiq_inline do
+      aggregate_failures do
+        expect(::Releases::CreateEvidenceService).not_to receive(:execute)
+        expect { described_class.new.perform }.to change(Releases::Evidence, :count).by(0)
+      end
+    end
+  end
+
+  context 'when `released_at` in inside the window' do
+    context 'when Evidence has not been created' do
+      let(:release) { create(:release, project: project, released_at: 1.hour.since) }
+
+      it 'creates a new Evidence record', :sidekiq_inline do
+        expect_next_instance_of(::Releases::CreateEvidenceService, release, { pipeline: nil }) do |service|
+          expect(service).to receive(:execute).and_call_original
+        end
+
+        expect { described_class.new.perform }.to change(Releases::Evidence, :count).by(1)
+      end
+    end
+
+    context 'when evidence has already been created' do
+      let(:release) { create(:release, project: project, released_at: 1.hour.since) }
+      let!(:evidence) { create(:evidence, release: release )}
+
+      it_behaves_like 'does not create a new Evidence record'
+    end
+  end
+
+  context 'when `released_at` is outside the window' do
+    let(:release) { create(:release, project: project, released_at: 300.minutes.since) }
+
+    it_behaves_like 'does not create a new Evidence record'
+  end
+end