1 files changed, 37 insertions, 27 deletions

diff --git a/app/services/ci/register_job_service.rb b/app/services/ci/register_job_service.rb
index 461647ffccc..6280bf4c986 100644
--- a/app/services/ci/register_job_service.rb
+++ b/app/services/ci/register_job_service.rb
@@ -22,11 +22,27 @@ module Ci
     end
 
     def execute(params = {})
+      db_all_caught_up = ::Gitlab::Database::LoadBalancing::Sticking.all_caught_up?(:runner, runner.id)
+
       @metrics.increment_queue_operation(:queue_attempt)
 
-      @metrics.observe_queue_time(:process, @runner.runner_type) do
+      result = @metrics.observe_queue_time(:process, @runner.runner_type) do
         process_queue(params)
       end
+
+      # Since we execute this query against replica it might lead to false-positive
+      # We might receive the positive response: "hi, we don't have any more builds for you".
+      # This might not be true. If our DB replica is not up-to date with when runner event was generated
+      # we might still have some CI builds to be picked. Instead we should say to runner:
+      # "Hi, we don't have any more builds now,  but not everything is right anyway, so try again".
+      # Runner will retry, but again, against replica, and again will check if replication lag did catch-up.
+      if !db_all_caught_up && !result.build
+        metrics.increment_queue_operation(:queue_replication_lag)
+
+        ::Ci::RegisterJobService::Result.new(nil, false) # rubocop:disable Cop/AvoidReturnFromBlocks
+      else
+        result
+      end
     end
 
     private
@@ -109,25 +125,23 @@ module Ci
         builds = builds.queued_before(params[:job_age].seconds.ago)
       end
 
-      if Feature.enabled?(:ci_register_job_service_one_by_one, runner, default_enabled: true)
-        build_ids = retrieve_queue(-> { builds.pluck(:id) })
-
-        @metrics.observe_queue_size(-> { build_ids.size }, @runner.runner_type)
+      build_ids = retrieve_queue(-> { builds.pluck(:id) })
 
-        build_ids.each do |build_id|
-          yield Ci::Build.find(build_id)
-        end
-      else
-        builds_array = retrieve_queue(-> { builds.to_a })
-
-        @metrics.observe_queue_size(-> { builds_array.size }, @runner.runner_type)
+      @metrics.observe_queue_size(-> { build_ids.size }, @runner.runner_type)
 
-        builds_array.each(&blk)
+      build_ids.each do |build_id|
+        yield Ci::Build.find(build_id)
       end
     end
     # rubocop: enable CodeReuse/ActiveRecord
 
     def retrieve_queue(queue_query_proc)
+      ##
+      # We want to reset a load balancing session to discard the side
+      # effects of writes that could have happened prior to this moment.
+      #
+      ::Gitlab::Database::LoadBalancing::Session.clear_session
+
       @metrics.observe_queue_time(:retrieve, @runner.runner_type) do
         queue_query_proc.call
       end
@@ -182,13 +196,7 @@ module Ci
     end
 
     def max_queue_depth
-      @max_queue_depth ||= begin
-        if Feature.enabled?(:gitlab_ci_builds_queue_limit, runner, default_enabled: true)
-          MAX_QUEUE_DEPTH
-        else
-          ::Gitlab::Database::MAX_INT_VALUE
-        end
-      end
+      MAX_QUEUE_DEPTH
     end
 
     # Force variables evaluation to occur now
@@ -271,15 +279,11 @@ module Ci
           .order(Arel.sql('COALESCE(project_builds.running_builds, 0) ASC'), 'ci_builds.id ASC')
       end
     end
-    # rubocop: enable CodeReuse/ActiveRecord
 
-    # rubocop: disable CodeReuse/ActiveRecord
     def builds_for_project_runner
       new_builds.where(project: runner.projects.without_deleted.with_builds_enabled).order('id ASC')
     end
-    # rubocop: enable CodeReuse/ActiveRecord
 
-    # rubocop: disable CodeReuse/ActiveRecord
     def builds_for_group_runner
       # Workaround for weird Rails bug, that makes `runner.groups.to_sql` to return `runner_id = NULL`
       groups = ::Group.joins(:runner_namespaces).merge(runner.runner_namespaces)
@@ -291,17 +295,23 @@ module Ci
         .without_deleted
       new_builds.where(project: projects).order('id ASC')
     end
-    # rubocop: enable CodeReuse/ActiveRecord
 
-    # rubocop: disable CodeReuse/ActiveRecord
     def running_builds_for_shared_runners
       Ci::Build.running.where(runner: Ci::Runner.instance_type)
         .group(:project_id).select(:project_id, 'count(*) AS running_builds')
     end
+
+    def all_builds
+      if Feature.enabled?(:ci_pending_builds_queue_join, runner, default_enabled: :yaml)
+        Ci::Build.joins(:queuing_entry)
+      else
+        Ci::Build.all
+      end
+    end
     # rubocop: enable CodeReuse/ActiveRecord
 
     def new_builds
-      builds = Ci::Build.pending.unstarted
+      builds = all_builds.pending.unstarted
       builds = builds.ref_protected if runner.ref_protected?
       builds
     end