Add GenericEnqueuer for consistent job priorities

johha · johha · commit 6749230bc522 · 2025-03-03T12:45:26.000+01:00
Some of the delayed jobs enqueue other delayed jobs. An example for this is an app delete job which enqueues other jobs like blobstore delete and buildpack cache cleanup.
If CAPI is configured with dedicated job priorities or when dynamic job priorities is enabled, those secondary jobs might end up with a higher priority then the primary job.
This might result in less important jobs like blobstore delete to have a higher priority than more important ones like service instance create.

This change introduces a `GenericEnqueuer` which follows the singleton pattern to ensure that always the same enqueuer instance is used in context of the current thread.
`GenericEnqueuer` will be initialized and destroyed in the CCJob wrapper and therefore every job which gets enqueued by `GenericEnqueuer.shared.enqueue(job)` will have the same priority.

Additionally the `enqueue` and `enqueue_pollable` methods allow an optional `priority_increment` which increases the priority value and thus makes the job less important.
With this we can give secondary jobs a lower priority then their parent useful for e.g. blobstore delete jobs.
If a job has a dedicated priority configured this priority will be added to the current priority and `priority_increment`, ensuring that the priorities of the primary job are correctly applied.

For reoccurring jobs the priority can be locked with the `preserve_priority` parameter which will ensure the current/previous priority is used again.

| Job Type               | Preserve Priority | Config Priority | Increment | Final Priority                       |
|------------------------|-------------------|-----------------|-----------|--------------------------------------|
| **Parent Job**         | ❌ No             | `100`           | `nil`     | **100**                              |
| **Sub-Job A**          | ❌ No             | `200`           | `50`      | **350** (100+200+50)                 |
| **Sub-Job B**          | ❌ No             | `nil`           | `50`      | **150** (100+50)                     |
| **Sub-Job C**          | ❌ No             | `nil`           | `nil`     | **100** (inherits parent)            |
| **Tertiary Job A1**    | ❌ No             | `50`            | `20`      | **420** (350+50+20)                  |
| **Tertiary Job B1**    | ✅ Yes            | `10` (ignored)  | `30`      | **150** (preserved from Sub-Job B)   |
| **Tertiary Job C1**    | ❌ No             | `nil`           | `50`      | **150** (100+50)                     |
| **Re-enqueued Job**    |  ✅ Yes           | `20` (ignored)  | `100`     | **42** (original preserved priority) |
diff --git a/app/actions/app_delete.rb b/app/actions/app_delete.rb
@@ -97,7 +97,7 @@ def route_mappings_to_delete(app)
 
     def delete_buildpack_cache(app)
       delete_job = Jobs::V3::BuildpackCacheDelete.new(app.guid)
-      Jobs::Enqueuer.new(queue: Jobs::Queues.generic).enqueue(delete_job)
+      Jobs::GenericEnqueuer.shared.enqueue(delete_job, priority_increment: Jobs::REDUCED_PRIORITY)
     end
 
     def logger
diff --git a/app/actions/buildpack_delete.rb b/app/actions/buildpack_delete.rb
@@ -8,7 +8,7 @@ def delete(buildpacks)
         end
         if buildpack.key
           blobstore_delete = Jobs::Runtime::BlobstoreDelete.new(buildpack.key, :buildpack_blobstore)
-          Jobs::Enqueuer.new(queue: Jobs::Queues.generic).enqueue(blobstore_delete)
+          Jobs::GenericEnqueuer.shared.enqueue(blobstore_delete, priority_increment: Jobs::REDUCED_PRIORITY)
         end
       end
 
diff --git a/app/actions/droplet_copy.rb b/app/actions/droplet_copy.rb
@@ -60,7 +60,7 @@ def copy_buildpack_droplet(new_droplet)
       new_droplet.buildpack_lifecycle_data(reload: true)
 
       copy_job = Jobs::V3::DropletBitsCopier.new(@source_droplet.guid, new_droplet.guid)
-      Jobs::Enqueuer.new(queue: Jobs::Queues.generic).enqueue(copy_job)
+      Jobs::GenericEnqueuer.shared.enqueue(copy_job)
     end
   end
 end
diff --git a/app/actions/droplet_delete.rb b/app/actions/droplet_delete.rb
@@ -16,7 +16,7 @@ def delete(droplets)
 
         if droplet.blobstore_key
           blobstore_delete = Jobs::Runtime::BlobstoreDelete.new(droplet.blobstore_key, :droplet_blobstore)
-          Jobs::Enqueuer.new(queue: Jobs::Queues.generic).enqueue(blobstore_delete)
+          Jobs::GenericEnqueuer.shared.enqueue(blobstore_delete, priority_increment: Jobs::REDUCED_PRIORITY)
         end
 
         Repositories::DropletEventRepository.record_delete(
diff --git a/app/actions/mixins/bindings_delete.rb b/app/actions/mixins/bindings_delete.rb
@@ -1,5 +1,6 @@
 require 'jobs/queues'
 require 'jobs/enqueuer'
+require 'jobs/generic_enqueuer'
 require 'jobs/v3/delete_binding_job'
 require 'jobs/v3/delete_service_binding_job_factory'
 
@@ -19,7 +20,7 @@ def delete_bindings(bindings, user_audit_info:)
           result = binding_delete_action.delete(binding)
           unless result[:finished]
             polling_job = DeleteBindingJob.new(type, binding.guid, user_audit_info:)
-            Jobs::Enqueuer.new(queue: Jobs::Queues.generic).enqueue_pollable(polling_job)
+            Jobs::GenericEnqueuer.shared.enqueue_pollable(polling_job)
             unbinding_operation_in_progress!(binding)
           end
         rescue StandardError => e
diff --git a/app/actions/package_copy.rb b/app/actions/package_copy.rb
@@ -20,7 +20,7 @@ def copy(destination_app_guid:, source_package:, user_audit_info:, record_event:
       package.db.transaction do
         package.save
 
-        @enqueued_job = Jobs::Enqueuer.new(queue: Jobs::Queues.generic).enqueue(Jobs::V3::PackageBitsCopier.new(source_package.guid, package.guid)) if source_package.type == 'bits'
+        @enqueued_job = Jobs::GenericEnqueuer.shared.enqueue(Jobs::V3::PackageBitsCopier.new(source_package.guid, package.guid)) if source_package.type == 'bits'
 
         record_audit_event(package, source_package, user_audit_info) if record_event
       end
diff --git a/app/actions/package_delete.rb b/app/actions/package_delete.rb
@@ -1,3 +1,5 @@
+require 'jobs/generic_enqueuer'
+
 module VCAP::CloudController
   class PackageDelete
     def initialize(user_audit_info)
@@ -10,7 +12,7 @@ def delete(packages)
       packages.each do |package|
         unless package.docker?
           package_src_delete_job = create_package_source_deletion_job(package)
-          Jobs::Enqueuer.new(queue: Jobs::Queues.generic).enqueue(package_src_delete_job) if package_src_delete_job
+          Jobs::GenericEnqueuer.shared.enqueue(package_src_delete_job, priority_increment: Jobs::REDUCED_PRIORITY) if package_src_delete_job
         end
 
         package.destroy
diff --git a/app/actions/routing/route_delete.rb b/app/actions/routing/route_delete.rb
@@ -15,7 +15,7 @@ def delete_sync(route:, recursive:)
     end
 
     def delete_async(route:, recursive:)
-      Jobs::Enqueuer.new(queue: Jobs::Queues.generic).enqueue(do_delete(recursive, route))
+      Jobs::GenericEnqueuer.shared.enqueue(do_delete(recursive, route))
     end
 
     def delete_unmapped_route(route:)
diff --git a/app/actions/service_broker_create.rb b/app/actions/service_broker_create.rb
@@ -31,7 +31,7 @@ def create(message)
           service_event_repository.record_broker_event_with_request(:create, broker, message.audit_hash)
 
           synchronization_job = SynchronizeBrokerCatalogJob.new(broker.guid, user_audit_info: service_event_repository.user_audit_info)
-          pollable_job = Jobs::Enqueuer.new(queue: Jobs::Queues.generic).enqueue_pollable(synchronization_job)
+          pollable_job = Jobs::GenericEnqueuer.shared.enqueue_pollable(synchronization_job)
         end
 
         { pollable_job: }
diff --git a/app/actions/services/locks/deleter_lock.rb b/app/actions/services/locks/deleter_lock.rb
@@ -49,7 +49,7 @@ def unlock_and_destroy!
 
     def enqueue_and_unlock!(attributes_to_update, job)
       service_instance.save_and_update_operation(attributes_to_update)
-      Jobs::Enqueuer.new(queue: Jobs::Queues.generic).enqueue(job)
+      Jobs::GenericEnqueuer.shared.enqueue(job)
       @needs_unlock = false
     end
 
diff --git a/app/actions/services/locks/updater_lock.rb b/app/actions/services/locks/updater_lock.rb
@@ -47,7 +47,7 @@ def asynchronous_unlock!
     end
 
     def enqueue_unlock!(job)
-      Jobs::Enqueuer.new(queue: Jobs::Queues.generic).enqueue(job)
+      Jobs::GenericEnqueuer.shared.enqueue(job)
       @needs_unlock = false
     end
 
diff --git a/app/actions/services/service_instance_create.rb b/app/actions/services/service_instance_create.rb
@@ -48,7 +48,7 @@ def setup_async_job(request_attrs, service_instance)
         @services_event_repository.user_audit_info,
         request_attrs
       )
-      Jobs::Enqueuer.new(queue: Jobs::Queues.generic).enqueue(job)
+      Jobs::GenericEnqueuer.shared.enqueue(job)
       @services_event_repository.record_service_instance_event(:start_create, service_instance, request_attrs)
     end
 
diff --git a/app/actions/space_delete.rb b/app/actions/space_delete.rb
@@ -56,7 +56,7 @@ def delete_service_instances(space_model)
         result = service_instance_deleter.delete
         unless result[:finished]
           polling_job = V3::DeleteServiceInstanceJob.new(service_instance.guid, @services_event_repository.user_audit_info)
-          Jobs::Enqueuer.new(queue: Jobs::Queues.generic).enqueue_pollable(polling_job)
+          Jobs::GenericEnqueuer.shared.enqueue_pollable(polling_job)
           errors << CloudController::Errors::ApiError.new_from_details('AsyncServiceInstanceOperationInProgress', service_instance.name)
         end
       rescue StandardError => e
diff --git a/app/actions/v2/services/service_binding_create.rb b/app/actions/v2/services/service_binding_create.rb
@@ -52,7 +52,7 @@ def create(app, service_instance, message, volume_mount_services_enabled, accept
 
           binding.save_with_new_operation({ type: 'create', state: 'in progress', broker_provided_operation: binding_result[:operation] })
           job = Jobs::Services::ServiceBindingStateFetch.new(binding.guid, @user_audit_info, message.audit_hash)
-          Jobs::Enqueuer.new(queue: Jobs::Queues.generic).enqueue(job)
+          Jobs::GenericEnqueuer.shared.enqueue(job)
           Repositories::ServiceBindingEventRepository.record_start_create(binding, @user_audit_info, message.audit_hash, manifest_triggered: @manifest_triggered)
         else
           binding.save
diff --git a/app/actions/v2/services/service_binding_delete.rb b/app/actions/v2/services/service_binding_delete.rb
@@ -19,7 +19,7 @@ def foreground_delete_request(service_binding)
     end
 
     def background_delete_request(service_binding)
-      Jobs::Enqueuer.new(queue: Jobs::Queues.generic).enqueue(Jobs::DeleteActionJob.new(ServiceBinding, service_binding.guid, self))
+      Jobs::GenericEnqueuer.shared.enqueue(Jobs::DeleteActionJob.new(ServiceBinding, service_binding.guid, self))
     end
 
     def delete(service_bindings)
@@ -39,7 +39,7 @@ def delete(service_bindings)
           service_binding.save_with_new_operation({ type: 'delete', state: 'in progress', broker_provided_operation: broker_response[:operation] })
 
           job = VCAP::CloudController::Jobs::Services::ServiceBindingStateFetch.new(service_binding.guid, @user_audit_info, {})
-          Jobs::Enqueuer.new(queue: Jobs::Queues.generic).enqueue(job)
+          Jobs::GenericEnqueuer.shared.enqueue(job)
           Repositories::ServiceBindingEventRepository.record_start_delete(service_binding, @user_audit_info)
         else
           service_binding.destroy
diff --git a/app/actions/v3/service_broker_update.rb b/app/actions/v3/service_broker_update.rb
@@ -58,7 +58,7 @@ def enqueue_update
             previous_broker_state,
             user_audit_info: service_event_repository.user_audit_info
           )
-          pollable_job = Jobs::Enqueuer.new(queue: Jobs::Queues.generic).enqueue_pollable(synchronization_job)
+          pollable_job = Jobs::GenericEnqueuer.shared.enqueue_pollable(synchronization_job)
         end
 
         pollable_job
diff --git a/app/actions/v3/service_instance_update_managed.rb b/app/actions/v3/service_instance_update_managed.rb
@@ -83,7 +83,7 @@ def enqueue_update
             user_audit_info: user_audit_info,
             audit_hash: message.audit_hash
           )
-          pollable_job = Jobs::Enqueuer.new(queue: Jobs::Queues.generic).enqueue_pollable(update_job)
+          pollable_job = Jobs::GenericEnqueuer.shared.enqueue_pollable(update_job)
           lock.asynchronous_unlock!
         ensure
           lock.unlock_and_fail! if lock.present? && lock.needs_unlock?
diff --git a/app/jobs/cc_job.rb b/app/jobs/cc_job.rb
@@ -1,6 +1,14 @@
 module VCAP::CloudController
   module Jobs
     class CCJob
+      def before(delayed_job)
+        GenericEnqueuer.shared(priority: delayed_job.priority)
+      end
+
+      def after(_delayed_job)
+        GenericEnqueuer.reset!
+      end
+
       def reschedule_at(time, attempts)
         time + (attempts**4) + 5
       end
diff --git a/app/jobs/delete_action_job.rb b/app/jobs/delete_action_job.rb
@@ -55,6 +55,9 @@ def delete_action_can_return_warnings?
         delete_action.respond_to?(:can_return_warnings?) && delete_action.can_return_warnings?
       end
 
+      # def delete_action_can_enqueue_jobs?
+      #   delete_action.respond_to?(:can_enqueue_jobs?) && delete_action.can_enqueue_jobs?
+      # end
       attr_reader :model_class, :delete_action
     end
   end
diff --git a/app/jobs/enqueuer.rb b/app/jobs/enqueuer.rb
@@ -16,16 +16,16 @@ def initialize(opts={})
         load_delayed_job_plugins
       end
 
-      def enqueue(job)
-        enqueue_job(job)
+      def enqueue(job, run_at: nil, priority_increment: nil)
+        enqueue_job(job, run_at:, priority_increment:)
       end
 
-      def enqueue_pollable(job, existing_guid: nil)
+      def enqueue_pollable(job, existing_guid: nil, run_at: nil, priority_increment: nil, preserve_priority: false)
         wrapped_job = PollableJobWrapper.new(job, existing_guid:)
 
         wrapped_job = yield wrapped_job if block_given?
 
-        delayed_job = enqueue_job(wrapped_job)
+        delayed_job = enqueue_job(wrapped_job, run_at:, priority_increment:, preserve_priority:)
         PollableJobModel.find_by_delayed_job(delayed_job)
       end
 
@@ -37,14 +37,26 @@ def run_inline(job)
 
       private
 
-      def enqueue_job(job)
+      def enqueue_job(job, run_at: nil, priority_increment: nil, preserve_priority: false)
         @opts['guid'] = SecureRandom.uuid
         request_id = ::VCAP::Request.current_id
         timeout_job = TimeoutJob.new(job, job_timeout(job))
         logging_context_job = LoggingContextJob.new(timeout_job, request_id)
-        job_priority = job_priority(job)
-        @opts[:priority] = job_priority unless @opts[:priority] || job_priority.nil?
-        Delayed::Job.enqueue(logging_context_job, @opts)
+
+        base_priority = @opts[:priority] || 0
+        priority_from_config = get_overwritten_job_priority_from_config(job) || 0
+
+        final_priority = base_priority
+        final_priority += priority_from_config unless preserve_priority
+        final_priority += [priority_increment, 0].max if priority_increment && !preserve_priority
+
+        local_opts = {}
+        # DelayedJob might have a different default priority. In the context of the enqueuer, we consider 0 as the default priority.
+        # Thus, we only set the priority if we use a non-default priority.
+        local_opts[:priority] = final_priority if final_priority > 0
+        local_opts[:run_at] = run_at if run_at
+
+        Delayed::Job.enqueue(logging_context_job, @opts.merge(local_opts))
       end
 
       def load_delayed_job_plugins
@@ -58,7 +70,7 @@ def job_timeout(job)
         @timeout_calculator.calculate(unwrapped_job.try(:job_name_in_configuration))
       end
 
-      def job_priority(job)
+      def get_overwritten_job_priority_from_config(job)
         unwrapped_job = unwrap_job(job)
         @priority_overwriter.get(unwrapped_job.try(:display_name)) ||
           @priority_overwriter.get(unwrapped_job.try(:job_name_in_configuration)) ||
diff --git a/app/jobs/generic_enqueuer.rb b/app/jobs/generic_enqueuer.rb
@@ -0,0 +1,20 @@
+module VCAP::CloudController
+  module Jobs
+    REDUCED_PRIORITY = 50
+
+    class GenericEnqueuer < Enqueuer
+      def self.shared(priority: nil)
+        stored_instance = Thread.current[:generic_enqueuer]
+        return stored_instance if stored_instance && priority.nil?
+
+        new_instance = new(queue: Jobs::Queues.generic, priority: priority)
+        Thread.current[:generic_enqueuer] ||= new_instance
+        new_instance
+      end
+
+      def self.reset!
+        Thread.current[:generic_enqueuer] = nil
+      end
+    end
+  end
+end
diff --git a/app/jobs/pollable_job_wrapper.rb b/app/jobs/pollable_job_wrapper.rb
@@ -27,7 +27,8 @@ def before_enqueue(job)
           user_guid = VCAP::CloudController::UserAuditInfo.from_context(VCAP::CloudController::SecurityContext).user_guid
 
           if VCAP::CloudController::Config.config.get(:jobs, :enable_dynamic_job_priorities) && user_guid
-            job.values[:priority] += PollableJobModel.number_of_active_jobs_by_user(user_guid)
+            # job.values[:priority] += PollableJobModel.number_of_active_jobs_by_user(user_guid)
+            job.values[:priority] += 100
           end
 
           PollableJobModel.create(
diff --git a/app/jobs/reoccurring_job.rb b/app/jobs/reoccurring_job.rb
@@ -13,7 +13,7 @@ def success(current_delayed_job)
         elsif next_enqueue_would_exceed_maximum_duration?
           expire!
         else
-          enqueue_next_job(pollable_job, current_delayed_job.priority)
+          enqueue_next_job(pollable_job)
         end
       end
 
@@ -75,15 +75,10 @@ def expire!
         raise CloudController::Errors::ApiError.new_from_details('JobTimeout')
       end
 
-      def enqueue_next_job(pollable_job, priority)
-        opts = {
-          queue: Jobs::Queues.generic,
-          run_at: Delayed::Job.db_time_now + next_execution_in,
-          priority: priority
-        }
-
+      def enqueue_next_job(pollable_job)
+        run_at = Delayed::Job.db_time_now + next_execution_in
         @retry_number += 1
-        Jobs::Enqueuer.new(opts).enqueue_pollable(self, existing_guid: pollable_job.guid)
+        Jobs::GenericEnqueuer.shared.enqueue_pollable(self, existing_guid: pollable_job.guid, run_at: run_at, preserve_priority: true)
       end
     end
   end
diff --git a/app/jobs/runtime/expired_blob_cleanup.rb b/app/jobs/runtime/expired_blob_cleanup.rb
@@ -23,11 +23,11 @@ def max_attempts
         end
 
         def enqueue_droplet_delete_job(droplet_guid)
-          Jobs::Enqueuer.new(queue: Jobs::Queues.generic).enqueue(Jobs::Runtime::DeleteExpiredDropletBlob.new(droplet_guid))
+          Jobs::GenericEnqueuer.shared.enqueue(Jobs::Runtime::DeleteExpiredDropletBlob.new(droplet_guid))
         end
 
         def enqueue_package_delete_job(package_guid)
-          Jobs::Enqueuer.new(queue: Jobs::Queues.generic).enqueue(Jobs::Runtime::DeleteExpiredPackageBlob.new(package_guid))
+          Jobs::GenericEnqueuer.shared.enqueue(Jobs::Runtime::DeleteExpiredPackageBlob.new(package_guid))
         end
 
         def logger
diff --git a/app/jobs/v2/services/asynchronous_operations.rb b/app/jobs/v2/services/asynchronous_operations.rb
@@ -23,9 +23,9 @@ def retry_job(retry_after_header: '')
         end
 
         def enqueue_again
-          opts = { queue: Jobs::Queues.generic, run_at: Delayed::Job.db_time_now + next_execution_in }
+          run_at_time = Delayed::Job.db_time_now + next_execution_in
           self.retry_number += 1
-          Jobs::Enqueuer.new(opts).enqueue(self)
+          Jobs::GenericEnqueuer.shared.enqueue(self, run_at: run_at_time)
         end
 
         def default_polling_exponential_backoff
diff --git a/lib/services/service_brokers/v2/orphan_mitigator.rb b/lib/services/service_brokers/v2/orphan_mitigator.rb
@@ -13,8 +13,7 @@ def cleanup_failed_provision(service_instance)
             service_instance.service_plan.guid
           )
 
-          opts = { queue: VCAP::CloudController::Jobs::Queues.generic, run_at: Delayed::Job.db_time_now }
-          VCAP::CloudController::Jobs::Enqueuer.new(opts).enqueue(orphan_deprovision_job)
+          VCAP::CloudController::Jobs::GenericEnqueuer.shared.enqueue(orphan_deprovision_job, run_at: Delayed::Job.db_time_now)
         end
 
         def cleanup_failed_bind(service_binding)
@@ -24,8 +23,7 @@ def cleanup_failed_bind(service_binding)
             binding_info
           )
 
-          opts = { queue: VCAP::CloudController::Jobs::Queues.generic, run_at: Delayed::Job.db_time_now }
-          VCAP::CloudController::Jobs::Enqueuer.new(opts).enqueue(unbind_job)
+          VCAP::CloudController::Jobs::GenericEnqueuer.shared.enqueue(unbind_job, run_at: Delayed::Job.db_time_now)
         end
 
         def cleanup_failed_key(service_key)
@@ -35,8 +33,7 @@ def cleanup_failed_key(service_key)
             service_key.service_instance.guid
           )
 
-          opts = { queue: VCAP::CloudController::Jobs::Queues.generic, run_at: Delayed::Job.db_time_now }
-          VCAP::CloudController::Jobs::Enqueuer.new(opts).enqueue(key_delete_job)
+          VCAP::CloudController::Jobs::GenericEnqueuer.shared.enqueue(key_delete_job, run_at: Delayed::Job.db_time_now)
         end
       end
     end
diff --git a/spec/unit/actions/buildpack_delete_spec.rb b/spec/unit/actions/buildpack_delete_spec.rb
@@ -36,9 +36,9 @@ module VCAP::CloudController
         it 'first deletes the database record and afterwards the blob' do
           expect(buildpack).to receive(:destroy).ordered
           expect(Jobs::Runtime::BlobstoreDelete).to receive(:new).ordered
-          enqueue_job_dbl = double('Jobs::Enqueuer')
-          expect(Jobs::Enqueuer).to receive(:new).and_return(enqueue_job_dbl).ordered
-          expect(enqueue_job_dbl).to receive(:enqueue).ordered
+          generic_enqueuer_dbl = double('Jobs::GenericEnqueuer')
+          expect(Jobs::GenericEnqueuer).to receive(:shared).and_return(generic_enqueuer_dbl).ordered
+          expect(generic_enqueuer_dbl).to receive(:enqueue).ordered
 
           buildpack_delete.delete([buildpack])
         end
diff --git a/spec/unit/jobs/cc_job_spec.rb b/spec/unit/jobs/cc_job_spec.rb
diff --git a/spec/unit/jobs/enqueuer_spec.rb b/spec/unit/jobs/enqueuer_spec.rb
diff --git a/spec/unit/jobs/generic_enqueuer_spec.rb b/spec/unit/jobs/generic_enqueuer_spec.rb
diff --git a/spec/unit/jobs/reoccurring_job_spec.rb b/spec/unit/jobs/reoccurring_job_spec.rb
diff --git a/spec/unit/lib/services/service_brokers/v2/orphan_mitigator_spec.rb b/spec/unit/lib/services/service_brokers/v2/orphan_mitigator_spec.rb
