Add capability to discard duplicate jobs with concurrency configuration

app/models/solid_queue/job.rb

@@ -10,19 +10,22 @@ class EnqueueError < StandardError; end
 
     class << self
       def enqueue_all(active_jobs)
-        active_jobs_by_job_id = active_jobs.index_by(&:job_id)
+        enqueued_jobs_count = 0
 
         transaction do
           jobs = create_all_from_active_jobs(active_jobs)
-          prepare_all_for_execution(jobs).tap do |enqueued_jobs|
-            enqueued_jobs.each do |enqueued_job|
-              active_jobs_by_job_id[enqueued_job.active_job_id].provider_job_id = enqueued_job.id
-              active_jobs_by_job_id[enqueued_job.active_job_id].successfully_enqueued = true
-            end
+          enqueued_jobs_by_active_job_id = prepare_all_for_execution(jobs).index_by(&:active_job_id)
+
+          active_jobs.each do |active_job|
+            job = enqueued_jobs_by_active_job_id[active_job.job_id]
+            active_job.provider_job_id = job&.id
+            active_job.successfully_enqueued = job.present?
           end
+
+          enqueued_jobs_count = enqueued_jobs_by_active_job_id.count
         end
 
-        active_jobs.count(&:successfully_enqueued?)
+        enqueued_jobs_count
       end
 
       def enqueue(active_job, scheduled_at: Time.current)
@@ -49,7 +52,7 @@ def create_from_active_job(active_job)
         def create_all_from_active_jobs(active_jobs)
           job_rows = active_jobs.map { |job| attributes_from_active_job(job) }
           insert_all(job_rows)
-          where(active_job_id: active_jobs.map(&:job_id))
+          where(active_job_id: active_jobs.map(&:job_id)).order(id: :asc)
         end
 
         def attributes_from_active_job(active_job)

app/models/solid_queue/job/concurrency_controls.rb

@@ -8,7 +8,7 @@ module ConcurrencyControls
       included do
         has_one :blocked_execution
 
-        delegate :concurrency_limit, :concurrency_duration, to: :job_class
+        delegate :concurrency_limit, :concurrency_at_limit, :concurrency_duration, to: :job_class
 
         before_destroy :unblock_next_blocked_job, if: -> { concurrency_limited? && ready? }
       end
@@ -34,6 +34,10 @@ def blocked?
       end
 
       private
+        def discard_concurrent?
+          concurrency_at_limit == :discard
+        end
+
         def acquire_concurrency_lock
           return true unless concurrency_limited?
 

app/models/solid_queue/job/executable.rb

@@ -66,6 +66,7 @@ def prepare_for_execution
 
       def dispatch
         if acquire_concurrency_lock then ready
+        elsif discard_concurrent? then discard
         else
           block
         end

lib/active_job/concurrency_controls.rb

@@ -11,15 +11,17 @@ module ConcurrencyControls
       class_attribute :concurrency_group, default: DEFAULT_CONCURRENCY_GROUP, instance_accessor: false
 
       class_attribute :concurrency_limit
+      class_attribute :concurrency_at_limit
       class_attribute :concurrency_duration, default: SolidQueue.default_concurrency_control_period
     end
 
     class_methods do
-      def limits_concurrency(key:, to: 1, group: DEFAULT_CONCURRENCY_GROUP, duration: SolidQueue.default_concurrency_control_period)
+      def limits_concurrency(key:, to: 1, group: DEFAULT_CONCURRENCY_GROUP, duration: SolidQueue.default_concurrency_control_period, at_limit: :block)
         self.concurrency_key = key
         self.concurrency_limit = to
         self.concurrency_group = group
         self.concurrency_duration = duration
+        self.concurrency_at_limit = at_limit
       end
     end
 

test/models/solid_queue/job_test.rb

@@ -10,6 +10,14 @@ def perform(job_result)
     end
   end
 
+  class DiscardedNonOverlappingJob < NonOverlappingJob
+    limits_concurrency key: ->(job_result, **) { job_result }, at_limit: :discard
+  end
+
+  class DiscardedOverlappingJob < NonOverlappingJob
+    limits_concurrency to: 2, key: ->(job_result, **) { job_result }, at_limit: :discard
+  end
+
   class NonOverlappingGroupedJob1 < NonOverlappingJob
     limits_concurrency key: ->(job_result, **) { job_result }, group: "MyGroup"
   end
@@ -98,6 +106,53 @@ class NonOverlappingGroupedJob2 < NonOverlappingJob
     assert_equal active_job.concurrency_key, job.concurrency_key
   end
 
+  test "enqueue jobs with discarding concurrency controls" do
+    assert_ready do
+      active_job = DiscardedNonOverlappingJob.perform_later(@result, name: "A")
+      assert_equal 1, active_job.concurrency_limit
+      assert_equal "SolidQueue::JobTest::DiscardedNonOverlappingJob/JobResult/#{@result.id}", active_job.concurrency_key
+    end
+
+    assert_discarded do
+      active_job = DiscardedNonOverlappingJob.perform_later(@result, name: "A")
+      assert_equal 1, active_job.concurrency_limit
+      assert_equal "SolidQueue::JobTest::DiscardedNonOverlappingJob/JobResult/#{@result.id}", active_job.concurrency_key
+    end
+  end
+
+  test "enqueuing multiple jobs with enqueue_all and concurrency controls" do
+    jobs = [
+      job_1 = DiscardedNonOverlappingJob.new(@result, name: "A"),
+      job_2 = DiscardedNonOverlappingJob.new(@result, name: "B")
+    ]
+
+    enqueued_jobs_count = SolidQueue::Job.enqueue_all(jobs)
+    assert_equal enqueued_jobs_count, 1
+
+    assert job_1.successfully_enqueued?
+    assert_not job_2.successfully_enqueued?
+  end
+
+  test "enqueue jobs with discarding concurrency controls when below limit" do
+    assert_ready do
+      active_job = DiscardedOverlappingJob.perform_later(@result, name: "A")
+      assert_equal 2, active_job.concurrency_limit
+      assert_equal "SolidQueue::JobTest::DiscardedOverlappingJob/JobResult/#{@result.id}", active_job.concurrency_key
+    end
+
+    assert_ready do
+      active_job = DiscardedOverlappingJob.perform_later(@result, name: "A")
+      assert_equal 2, active_job.concurrency_limit
+      assert_equal "SolidQueue::JobTest::DiscardedOverlappingJob/JobResult/#{@result.id}", active_job.concurrency_key
+    end
+
+    assert_discarded do
+      active_job = DiscardedOverlappingJob.perform_later(@result, name: "A")
+      assert_equal 2, active_job.concurrency_limit
+      assert_equal "SolidQueue::JobTest::DiscardedOverlappingJob/JobResult/#{@result.id}", active_job.concurrency_key
+    end
+  end
+
   test "enqueue jobs with concurrency controls in the same concurrency group" do
     assert_ready do
       active_job = NonOverlappingGroupedJob1.perform_later(@result, name: "A")
@@ -289,8 +344,12 @@ def assert_blocked(&block)
       assert SolidQueue::Job.last.blocked?
     end
 
-    def assert_job_counts(ready: 0, scheduled: 0, blocked: 0, &block)
-      assert_difference -> { SolidQueue::Job.count }, +(ready + scheduled + blocked) do
+    def assert_discarded(&block)
+      assert_job_counts(discarded: 1, &block)
+    end
+
+    def assert_job_counts(ready: 0, scheduled: 0, blocked: 0, discarded: 0, &block)
+      assert_difference -> { SolidQueue::Job.count }, +(ready + scheduled + blocked + discarded) do
         assert_difference -> { SolidQueue::ReadyExecution.count }, +ready do
           assert_difference -> { SolidQueue::ScheduledExecution.count }, +scheduled do
             assert_difference -> { SolidQueue::BlockedExecution.count }, +blocked, &block