Merge branch 'process-commits-using-sidekiq' into 'master'

Processing Commits Using Sidekiq

This moves the code  of `GitPushService#process_commit_messages` into a separate Sidekiq worker. This allows processing of commits to happen in parallel, speeding up the process. See the individual commit (messages) for more information.

Part of https://gitlab.com/gitlab-org/gitlab-ce/issues/15463

See merge request !6802

app/models/concerns/issuable.rb

@@ -286,6 +286,11 @@ module Issuable
     false
   end
 
+  def assignee_or_author?(user)
+    # We're comparing IDs here so we don't need to load any associations.
+    author_id == user.id || assignee_id == user.id
+  end
+
   def record_metrics
     metrics = self.metrics || create_metrics
     metrics.record!

app/models/external_issue.rb

@@ -29,6 +29,15 @@ class ExternalIssue
     @project
   end
 
+  def project_id
+    @project.id
+  end
+
+  # Pattern used to extract `JIRA-123` issue references from text
+  def self.reference_pattern
+    @reference_pattern ||= %r{(?<issue>\b([A-Z][A-Z0-9_]+-)\d+)}
+  end
+
   def to_reference(_from_project = nil)
     id
   end

app/models/issue_collection.rb

+# IssueCollection can be used to reduce a list of issues down to a subset.
+#
+# IssueCollection is not meant to be some sort of Enumerable, instead it's meant
+# to take a list of issues and return a new list of issues based on some
+# criteria. For example, given a list of issues you may want to return a list of
+# issues that can be read or updated by a given user.
+class IssueCollection
+  attr_reader :collection
+
+  def initialize(collection)
+    @collection = collection
+  end
+
+  # Returns all the issues that can be updated by the user.
+  def updatable_by_user(user)
+    return collection if user.admin?
+
+    # Given all the issue projects we get a list of projects that the current
+    # user has at least reporter access to.
+    projects_with_reporter_access = user.
+      projects_with_reporter_access_limited_to(project_ids).
+      pluck(:id)
+
+    collection.select do |issue|
+      if projects_with_reporter_access.include?(issue.project_id)
+        true
+      elsif issue.is_a?(Issue)
+        issue.assignee_or_author?(user)
+      else
+        false
+      end
+    end
+  end
+
+  alias_method :visible_to, :updatable_by_user
+
+  private
+
+  def project_ids
+    @project_ids ||= collection.map(&:project_id).uniq
+  end
+end

app/models/user.rb

@@ -444,6 +444,16 @@ class User < ActiveRecord::Base
     Project.where("projects.id IN (#{projects_union(min_access_level).to_sql})")
   end
 
+  # Returns the projects this user has reporter (or greater) access to, limited
+  # to at most the given projects.
+  #
+  # This method is useful when you have a list of projects and want to
+  # efficiently check to which of these projects the user has at least reporter
+  # access.
+  def projects_with_reporter_access_limited_to(projects)
+    authorized_projects(Gitlab::Access::REPORTER).where(id: projects)
+  end
+
   def viewable_starred_projects
     starred_projects.where("projects.visibility_level IN (?) OR projects.id IN (#{projects_union.to_sql})",
                            [Project::PUBLIC, Project::INTERNAL])

app/policies/issuable_policy.rb

@@ -4,7 +4,7 @@ class IssuablePolicy < BasePolicy
   end
 
   def rules
-    if @user && (@subject.author == @user || @subject.assignee == @user)
+    if @user && @subject.assignee_or_author?(@user)
       can! :"read_#{action_name}"
       can! :"update_#{action_name}"
     end

app/policies/issue_policy.rb

@@ -8,9 +8,8 @@ class IssuePolicy < IssuablePolicy
 
     if @subject.confidential? && !can_read_confidential?
       cannot! :read_issue
-      cannot! :admin_issue
       cannot! :update_issue
-      cannot! :read_issue
+      cannot! :admin_issue
     end
   end
 
@@ -18,11 +17,7 @@ class IssuePolicy < IssuablePolicy
 
   def can_read_confidential?
     return false unless @user
-    return true if @user.admin?
-    return true if @subject.author == @user
-    return true if @subject.assignee == @user
-    return true if @subject.project.team.member?(@user, Gitlab::Access::REPORTER)
 
-    false
+    IssueCollection.new([@subject]).visible_to(@user).any?
   end
 end

app/services/git_push_service.rb

@@ -105,35 +105,11 @@ class GitPushService < BaseService
   # Extract any GFM references from the pushed commit messages. If the configured issue-closing regex is matched,
   # close the referenced Issue. Create cross-reference Notes corresponding to any other referenced Mentionables.
   def process_commit_messages
-    is_default_branch = is_default_branch?
-
-    authors = Hash.new do |hash, commit|
-      email = commit.author_email
-      next hash[email] if hash.has_key?(email)
-
-      hash[email] = commit_user(commit)
-    end
+    default = is_default_branch?
 
     @push_commits.each do |commit|
-      # Keep track of the issues that will be actually closed because they are on a default branch.
-      # Hence, when creating cross-reference notes, the not-closed issues (on non-default branches)
-      # will also have cross-reference.
-      closed_issues = []
-
-      if is_default_branch
-        # Close issues if these commits were pushed to the project's default branch and the commit message matches the
-        # closing regex. Exclude any mentioned Issues from cross-referencing even if the commits are being pushed to
-        # a different branch.
-        closed_issues = commit.closes_issues(current_user)
-        closed_issues.each do |issue|
-          if can?(current_user, :update_issue, issue)
-            Issues::CloseService.new(project, authors[commit], {}).execute(issue, commit: commit)
-          end
-        end
-      end
-
-      commit.create_cross_references!(authors[commit], closed_issues)
-      update_issue_metrics(commit, authors)
+      ProcessCommitWorker.
+        perform_async(project.id, current_user.id, commit.id, default)
     end
   end
 
@@ -176,11 +152,4 @@ class GitPushService < BaseService
   def branch_name
     @branch_name ||= Gitlab::Git.ref_name(params[:ref])
   end
-
-  def update_issue_metrics(commit, authors)
-    mentioned_issues = commit.all_references(authors[commit]).issues
-
-    Issue::Metrics.where(issue_id: mentioned_issues.map(&:id), first_mentioned_in_commit_at: nil).
-      update_all(first_mentioned_in_commit_at: commit.committed_date)
-  end
 end

app/services/issues/close_service.rb

 module Issues
   class CloseService < Issues::BaseService
+    # Closes the supplied issue if the current user is able to do so.
     def execute(issue, commit: nil, notifications: true, system_note: true)
       return issue unless can?(current_user, :update_issue, issue)
 
+      close_issue(issue,
+                  commit: commit,
+                  notifications: notifications,
+                  system_note: system_note)
+    end
+
+    # Closes the supplied issue without checking if the user is authorized to
+    # do so.
+    #
+    # The code calling this method is responsible for ensuring that a user is
+    # allowed to close the given issue.
+    def close_issue(issue, commit: nil, notifications: true, system_note: true)
       if project.jira_tracker? && project.jira_service.active
         project.jira_service.execute(commit, issue)
         todo_service.close_issue(issue, current_user)

app/workers/process_commit_worker.rb

+# Worker for processing individiual commit messages pushed to a repository.
+#
+# Jobs for this worker are scheduled for every commit that is being pushed. As a
+# result of this the workload of this worker should be kept to a bare minimum.
+# Consider using an extra worker if you need to add any extra (and potentially
+# slow) processing of commits.
+class ProcessCommitWorker
+  include Sidekiq::Worker
+  include DedicatedSidekiqQueue
+
+  # project_id - The ID of the project this commit belongs to.
+  # user_id - The ID of the user that pushed the commit.
+  # commit_sha - The SHA1 of the commit to process.
+  # default - The data was pushed to the default branch.
+  def perform(project_id, user_id, commit_sha, default = false)
+    project = Project.find_by(id: project_id)
+
+    return unless project
+
+    user = User.find_by(id: user_id)
+
+    return unless user
+
+    commit = find_commit(project, commit_sha)
+
+    return unless commit
+
+    author = commit.author || user
+
+    process_commit_message(project, commit, user, author, default)
+
+    update_issue_metrics(commit, author)
+  end
+
+  def process_commit_message(project, commit, user, author, default = false)
+    closed_issues = default ? commit.closes_issues(user) : []
+
+    unless closed_issues.empty?
+      close_issues(project, user, author, commit, closed_issues)
+    end
+
+    commit.create_cross_references!(author, closed_issues)
+  end
+
+  def close_issues(project, user, author, commit, issues)
+    # We don't want to run permission related queries for every single issue,
+    # therefor we use IssueCollection here and skip the authorization check in
+    # Issues::CloseService#execute.
+    IssueCollection.new(issues).updatable_by_user(user).each do |issue|
+      Issues::CloseService.new(project, author).
+        close_issue(issue, commit: commit)
+    end
+  end
+
+  def update_issue_metrics(commit, author)
+    mentioned_issues = commit.all_references(author).issues
+
+    Issue::Metrics.where(issue_id: mentioned_issues.map(&:id), first_mentioned_in_commit_at: nil).
+      update_all(first_mentioned_in_commit_at: commit.committed_date)
+  end
+
+  private
+
+  def find_commit(project, sha)
+    project.commit(sha)
+  end
+end

changelogs/unreleased/process-commits-using-sidekiq.yml

+---
+title: Process commits using a dedicated Sidekiq worker
+merge_request: 6802
+author: 

config/sidekiq_queues.yml

@@ -21,6 +21,7 @@
   - [post_receive, 5]
   - [merge, 5]
   - [update_merge_requests, 3]
+  - [process_commit, 2]
   - [new_note, 2]
   - [build, 2]
   - [pipeline, 2]

spec/models/concerns/issuable_spec.rb

@@ -341,4 +341,25 @@ describe Issue, "Issuable" do
       expect(Issue.with_label([bug.title, enhancement.title])).to match_array([issue2])
     end
   end
+
+  describe '#assignee_or_author?' do
+    let(:user) { build(:user, id: 1) }
+    let(:issue) { build(:issue) }
+
+    it 'returns true for a user that is assigned to an issue' do
+      issue.assignee = user
+
+      expect(issue.assignee_or_author?(user)).to eq(true)
+    end
+
+    it 'returns true for a user that is the author of an issue' do
+      issue.author = user
+
+      expect(issue.assignee_or_author?(user)).to eq(true)
+    end
+
+    it 'returns false for a user that is not the assignee or author' do
+      expect(issue.assignee_or_author?(user)).to eq(false)
+    end
+  end
 end

spec/models/external_issue_spec.rb

 require 'spec_helper'
 
 describe ExternalIssue, models: true do
-  let(:project) { double('project', to_reference: 'namespace1/project1') }
+  let(:project) { double('project', id: 1, to_reference: 'namespace1/project1') }
   let(:issue)   { described_class.new('EXT-1234', project) }
 
   describe 'modules' do
@@ -36,4 +36,10 @@ describe ExternalIssue, models: true do
       end
     end
   end
+
+  describe '#project_id' do
+    it 'returns the ID of the project' do
+      expect(issue.project_id).to eq(project.id)
+    end
+  end
 end

spec/models/issue_collection_spec.rb

+require 'spec_helper'
+
+describe IssueCollection do
+  let(:user) { create(:user) }
+  let(:project) { create(:project) }
+  let(:issue1) { create(:issue, project: project) }
+  let(:issue2) { create(:issue, project: project) }
+  let(:collection) { described_class.new([issue1, issue2]) }
+
+  describe '#collection' do
+    it 'returns the issues in the same order as the input Array' do
+      expect(collection.collection).to eq([issue1, issue2])
+    end
+  end
+
+  describe '#updatable_by_user' do
+    context 'using an admin user' do
+      it 'returns all issues' do
+        user = create(:admin)
+
+        expect(collection.updatable_by_user(user)).to eq([issue1, issue2])
+      end
+    end
+
+    context 'using a user that has no access to the project' do
+      it 'returns no issues when the user is not an assignee or author' do
+        expect(collection.updatable_by_user(user)).to be_empty
+      end
+
+      it 'returns the issues the user is assigned to' do
+        issue1.assignee = user
+
+        expect(collection.updatable_by_user(user)).to eq([issue1])
+      end
+
+      it 'returns the issues for which the user is the author' do
+        issue1.author = user
+
+        expect(collection.updatable_by_user(user)).to eq([issue1])
+      end
+    end
+
+    context 'using a user that has reporter access to the project' do
+      it 'returns the issues of the project' do
+        project.team << [user, :reporter]
+
+        expect(collection.updatable_by_user(user)).to eq([issue1, issue2])
+      end
+    end
+
+    context 'using a user that is the owner of a project' do
+      it 'returns the issues of the project' do
+        expect(collection.updatable_by_user(project.namespace.owner)).
+          to eq([issue1, issue2])
+      end
+    end
+  end
+
+  describe '#visible_to' do
+    it 'is an alias for updatable_by_user' do
+      updatable_by_user = described_class.instance_method(:updatable_by_user)
+      visible_to = described_class.instance_method(:visible_to)
+
+      expect(visible_to).to eq(updatable_by_user)
+    end
+  end
+end

spec/models/user_spec.rb

@@ -1205,4 +1205,40 @@ describe User, models: true do
       expect(user.viewable_starred_projects).not_to include(private_project)
     end
   end
+
+  describe '#projects_with_reporter_access_limited_to' do
+    let(:project1) { create(:project) }
+    let(:project2) { create(:project) }
+    let(:user) { create(:user) }
+
+    before do
+      project1.team << [user, :reporter]
+      project2.team << [user, :guest]
+    end
+
+    it 'returns the projects when using a single project ID' do
+      projects = user.projects_with_reporter_access_limited_to(project1.id)
+
+      expect(projects).to eq([project1])
+    end
+
+    it 'returns the projects when using an Array of project IDs' do
+      projects = user.projects_with_reporter_access_limited_to([project1.id])
+
+      expect(projects).to eq([project1])
+    end
+
+    it 'returns the projects when using an ActiveRecord relation' do
+      projects = user.
+        projects_with_reporter_access_limited_to(Project.select(:id))
+
+      expect(projects).to eq([project1])
+    end
+
+    it 'does not return projects you do not have reporter access to' do
+      projects = user.projects_with_reporter_access_limited_to(project2.id)
+
+      expect(projects).to be_empty
+    end
+  end
 end

spec/policies/issue_policy_spec.rb

+require 'spec_helper'
+
+describe IssuePolicy, models: true do
+  let(:user) { create(:user) }
+
+  describe '#rules' do
+    context 'using a regular issue' do
+      let(:project) { create(:project, :public) }
+      let(:issue) { create(:issue, project: project) }
+      let(:policies) { described_class.abilities(user, issue).to_set }
+
+      context 'with a regular user' do
+        it 'includes the read_issue permission' do
+          expect(policies).to include(:read_issue)
+        end
+
+        it 'does not include the admin_issue permission' do
+          expect(policies).not_to include(:admin_issue)
+        end
+
+        it 'does not include the update_issue permission' do
+          expect(policies).not_to include(:update_issue)
+        end
+      end
+
+      context 'with a user that is a project reporter' do
+        before do
+          project.team << [user, :reporter]
+        end
+
+        it 'includes the read_issue permission' do
+          expect(policies).to include(:read_issue)
+        end
+
+        it 'includes the admin_issue permission' do
+          expect(policies).to include(:admin_issue)
+        end
+
+        it 'includes the update_issue permission' do
+          expect(policies).to include(:update_issue)
+        end
+      end
+
+      context 'with a user that is a project guest' do
+        before do
+          project.team << [user, :guest]
+        end
+
+        it 'includes the read_issue permission' do
+          expect(policies).to include(:read_issue)
+        end
+
+        it 'does not include the admin_issue permission' do
+          expect(policies).not_to include(:admin_issue)
+        end
+
+        it 'does not include the update_issue permission' do
+          expect(policies).not_to include(:update_issue)
+        end
+      end
+    end
+
+    context 'using a confidential issue' do
+      let(:issue) { create(:issue, :confidential) }
+
+      context 'with a regular user' do
+        let(:policies) { described_class.abilities(user, issue).to_set }
+
+        it 'does not include the read_issue permission' do
+          expect(policies).not_to include(:read_issue)
+        end
+
+        it 'does not include the admin_issue permission' do
+          expect(policies).not_to include(:admin_issue)
+        end
+
+        it 'does not include the update_issue permission' do
+          expect(policies).not_to include(:update_issue)
+        end
+      end
+
+      context 'with a user that is a project member' do
+        let(:policies) { described_class.abilities(user, issue).to_set }
+
+        before do
+          issue.project.team << [user, :reporter]
+        end
+
+        it 'includes the read_issue permission' do
+          expect(policies).to include(:read_issue)
+        end
+
+        it 'includes the admin_issue permission' do
+          expect(policies).to include(:admin_issue)
+        end
+
+        it 'includes the update_issue permission' do
+          expect(policies).to include(:update_issue)
+        end
+      end
+
+      context 'without a user' do
+        let(:policies) { described_class.abilities(nil, issue).to_set }
+
+        it 'does not include the read_issue permission' do
+          expect(policies).not_to include(:read_issue)
+        end
+
+        it 'does not include the admin_issue permission' do
+          expect(policies).not_to include(:admin_issue)
+        end
+
+        it 'does not include the update_issue permission' do
+          expect(policies).not_to include(:update_issue)
+        end
+      end
+    end
+  end
+end

spec/services/git_push_service_spec.rb

@@ -302,6 +302,9 @@ describe GitPushService, services: true do
         author_email: commit_author.email
       )
 
+      allow_any_instance_of(ProcessCommitWorker).to receive(:find_commit).
+        and_return(commit)
+
       allow(project.repository).to receive(:commits_between).and_return([commit])
     end
 
@@ -357,6 +360,9 @@ describe GitPushService, services: true do
         committed_date: commit_time
       )
 
+      allow_any_instance_of(ProcessCommitWorker).to receive(:find_commit).
+        and_return(commit)
+
       allow(project.repository).to receive(:commits_between).and_return([commit])
     end
 
@@ -393,6 +399,9 @@ describe GitPushService, services: true do
       allow(project.repository).to receive(:commits_between).
         and_return([closing_commit])
 
+      allow_any_instance_of(ProcessCommitWorker).to receive(:find_commit).
+        and_return(closing_commit)
+
       project.team << [commit_author, :master]
     end
 
@@ -538,9 +547,16 @@ describe GitPushService, services: true do
     let(:housekeeping) { Projects::HousekeepingService.new(project) }
 
     before do
+      # Flush any raw Redis data stored by the housekeeping code.
+      Gitlab::Redis.with { |conn| conn.flushall }
+
       allow(Projects::HousekeepingService).to receive(:new).and_return(housekeeping)
     end
 
+    after do
+      Gitlab::Redis.with { |conn| conn.flushall }
+    end
+
     it 'does not perform housekeeping when not needed' do
       expect(housekeeping).not_to receive(:execute)
 

spec/services/issues/close_service_spec.rb

@@ -15,10 +15,39 @@ describe Issues::CloseService, services: true do
   end
 
   describe '#execute' do
+    let(:service) { described_class.new(project, user) }
+
+    it 'checks if the user is authorized to update the issue' do
+      expect(service).to receive(:can?).with(user, :update_issue, issue).
+        and_call_original
+
+      service.execute(issue)
+    end
+
+    it 'does not close the issue when the user is not authorized to do so' do
+      allow(service).to receive(:can?).with(user, :update_issue, issue).
+        and_return(false)
+
+      expect(service).not_to receive(:close_issue)
+      expect(service.execute(issue)).to eq(issue)
+    end
+
+    it 'closes the issue when the user is authorized to do so' do
+      allow(service).to receive(:can?).with(user, :update_issue, issue).
+        and_return(true)
+
+      expect(service).to receive(:close_issue).
+        with(issue, commit: nil, notifications: true, system_note: true)
+
+      service.execute(issue)
+    end
+  end
+
+  describe '#close_issue' do
     context "valid params" do
       before do
         perform_enqueued_jobs do
-          described_class.new(project, user).execute(issue)
+          described_class.new(project, user).close_issue(issue)
         end
       end
 
@@ -41,24 +70,12 @@ describe Issues::CloseService, services: true do
       end
     end
 
-    context 'current user is not authorized to close issue' do
-      before do
-        perform_enqueued_jobs do
-          described_class.new(project, guest).execute(issue)
-        end
-      end
-
-      it 'does not close the issue' do
-        expect(issue).to be_open
-      end
-    end
-
     context 'when issue is not confidential' do
       it 'executes issue hooks' do
         expect(project).to receive(:execute_hooks).with(an_instance_of(Hash), :issue_hooks)
         expect(project).to receive(:execute_services).with(an_instance_of(Hash), :issue_hooks)
 
-        described_class.new(project, user).execute(issue)
+        described_class.new(project, user).close_issue(issue)
       end
     end
 
@@ -69,14 +86,14 @@ describe Issues::CloseService, services: true do
         expect(project).to receive(:execute_hooks).with(an_instance_of(Hash), :confidential_issue_hooks)
         expect(project).to receive(:execute_services).with(an_instance_of(Hash), :confidential_issue_hooks)
 
-        described_class.new(project, user).execute(issue)
+        described_class.new(project, user).close_issue(issue)
       end
     end
 
     context 'external issue tracker' do
       before do
         allow(project).to receive(:default_issues_tracker?).and_return(false)
-        described_class.new(project, user).execute(issue)
+        described_class.new(project, user).close_issue(issue)
       end
 
       it { expect(issue).to be_valid }

spec/workers/process_commit_worker_spec.rb

+require 'spec_helper'
+
+describe ProcessCommitWorker do
+  let(:worker) { described_class.new }
+  let(:user) { create(:user) }
+  let(:project) { create(:project, :public) }
+  let(:issue) { create(:issue, project: project, author: user) }
+  let(:commit) { project.commit }
+
+  describe '#perform' do
+    it 'does not process the commit when the project does not exist' do
+      expect(worker).not_to receive(:close_issues)
+
+      worker.perform(-1, user.id, commit.id)
+    end
+
+    it 'does not process the commit when the user does not exist' do
+      expect(worker).not_to receive(:close_issues)
+
+      worker.perform(project.id, -1, commit.id)
+    end
+
+    it 'does not process the commit when the commit no longer exists' do
+      expect(worker).not_to receive(:close_issues)
+
+      worker.perform(project.id, user.id, 'this-should-does-not-exist')
+    end
+
+    it 'processes the commit message' do
+      expect(worker).to receive(:process_commit_message).and_call_original
+
+      worker.perform(project.id, user.id, commit.id)
+    end
+
+    it 'updates the issue metrics' do
+      expect(worker).to receive(:update_issue_metrics).and_call_original
+
+      worker.perform(project.id, user.id, commit.id)
+    end
+  end
+
+  describe '#process_commit_message' do
+    context 'when pushing to the default branch' do
+      it 'closes issues that should be closed per the commit message' do
+        allow(commit).to receive(:safe_message).
+          and_return("Closes #{issue.to_reference}")
+
+        expect(worker).to receive(:close_issues).
+          with(project, user, user, commit, [issue])
+
+        worker.process_commit_message(project, commit, user, user, true)
+      end
+    end
+
+    context 'when pushing to a non-default branch' do
+      it 'does not close any issues' do
+        allow(commit).to receive(:safe_message).
+          and_return("Closes #{issue.to_reference}")
+
+        expect(worker).not_to receive(:close_issues)
+
+        worker.process_commit_message(project, commit, user, user, false)
+      end
+    end
+
+    it 'creates cross references' do
+      expect(commit).to receive(:create_cross_references!)
+
+      worker.process_commit_message(project, commit, user, user)
+    end
+  end
+
+  describe '#close_issues' do
+    context 'when the user can update the issues' do
+      it 'closes the issues' do
+        worker.close_issues(project, user, user, commit, [issue])
+
+        issue.reload
+
+        expect(issue.closed?).to eq(true)
+      end
+    end
+
+    context 'when the user can not update the issues' do
+      it 'does not close the issues' do
+        other_user = create(:user)
+
+        worker.close_issues(project, other_user, other_user, commit, [issue])
+
+        issue.reload
+
+        expect(issue.closed?).to eq(false)
+      end
+    end
+  end
+
+  describe '#update_issue_metrics' do
+    it 'updates any existing issue metrics' do
+      allow(commit).to receive(:safe_message).
+        and_return("Closes #{issue.to_reference}")
+
+      worker.update_issue_metrics(commit, user)
+
+      metric = Issue::Metrics.first
+
+      expect(metric.first_mentioned_in_commit_at).to eq(commit.committed_date)
+    end
+  end
+end