Tidy up after merging two branches implementing the same feature

Brought together #523 and #586, and took some parts from each.
Clean and enhance the README as well.

Author: rosa

README.md

@@ -428,25 +428,25 @@ In the case of recurring tasks, if such error is raised when enqueuing the job c
 
 ## Concurrency controls
 
-Solid Queue extends Active Job with concurrency controls, that allows you to limit how many jobs of a certain type or with certain arguments can run at the same time. When limited in this way, jobs will be blocked from running, and they'll stay blocked until another job finishes and unblocks them, or after the set expiry time (concurrency limit's _duration_) elapses. Jobs can be configured to either be discarded or blocked.
+Solid Queue extends Active Job with concurrency controls, that allows you to limit how many jobs of a certain type or with certain arguments can run at the same time. When limited in this way, by default, jobs will be blocked from running, and they'll stay blocked until another job finishes and unblocks them, or after the set expiry time (concurrency limit's _duration_) elapses. Alternatively, jobs can be configured to be discarded instead of blocked. This means that if a job with certain arguments has already been enqueued, other jobs with the same characteristics (in the same concurrency _class_) won't be enqueued, they'll silently complete.
 
 ```ruby
 class MyJob < ApplicationJob
-  limits_concurrency to: max_concurrent_executions, key: ->(arg1, arg2, **) { ... }, duration: max_interval_to_guarantee_concurrency_limit, group: concurrency_group, on_conflict: conflict_behaviour
+  limits_concurrency to: max_concurrent_executions, key: ->(arg1, arg2, **) { ... }, duration: max_interval_to_guarantee_concurrency_limit, group: concurrency_group, on_conflict: on_conflict_behaviour
 
   # ...
 ```
 - `key` is the only required parameter, and it can be a symbol, a string or a proc that receives the job arguments as parameters and will be used to identify the jobs that need to be limited together. If the proc returns an Active Record record, the key will be built from its class name and `id`.
 - `to` is `1` by default.
 - `duration` is set to `SolidQueue.default_concurrency_control_period` by default, which itself defaults to `3 minutes`, but that you can configure as well.
 - `group` is used to control the concurrency of different job classes together. It defaults to the job class name.
-- `on_conflict` controls behaviour when enqueuing a job which is above the max concurrent executions for your configuration. 
-  - (default) `:block`; the job is blocked and is dispatched until another job completes and unblocks it
-  - `:discard`; the job is discarded
+- `on_conflict` controls behaviour when enqueuing a job that conflicts with the concurrency limits configured. It can be set to one of the following:
+  - (default) `:block`: the job is blocked and is dispatched when another job completes and unblocks it, or when the duration expires.
+  - `:discard`: the job is discarded (silently finishes). When you choose this option, bear in mind that if a job runs and fails to remove the concurrency lock (or _semaphore_, read below to know more about this), all jobs conflicting with it will be discarded up to the interval defined by `duration` has elapsed.
 
 When a job includes these controls, we'll ensure that, at most, the number of jobs (indicated as `to`) that yield the same `key` will be performed concurrently, and this guarantee will last for `duration` for each job enqueued. Note that there's no guarantee about _the order of execution_, only about jobs being performed at the same time (overlapping).
 
-The concurrency limits use the concept of semaphores when enqueuing, and work as follows: when a job is enqueued, we check if it specifies concurrency controls. If it does, we check the semaphore for the computed concurrency key. If the semaphore is open, we claim it and we set the job as _ready_. Ready means it can be picked up by workers for execution. When the job finishes executing (be it successfully or unsuccessfully, resulting in a failed execution), we signal the semaphore and try to unblock the next job with the same key, if any. Unblocking the next job doesn't mean running that job right away, but moving it from _blocked_ to _ready_. Since something can happen that prevents the first job from releasing the semaphore and unblocking the next job (for example, someone pulling a plug in the machine where the worker is running), we have the `duration` as a failsafe. Jobs that have been blocked for more than duration are candidates to be released, but only as many of them as the concurrency rules allow, as each one would need to go through the semaphore dance check. This means that the `duration` is not really about the job that's enqueued or being run, it's about the jobs that are blocked waiting. It's important to note that after one or more candidate jobs are unblocked (either because a job finishes or because `duration` expires and a semaphore is released), the `duration` timer for the still blocked jobs is reset. This happens indirectly via the expiration time of the semaphore, which is updated.
+The concurrency limits use the concept of semaphores when enqueuing, and work as follows: when a job is enqueued, we check if it specifies concurrency controls. If it does, we check the semaphore for the computed concurrency key. If the semaphore is open, we claim it and we set the job as _ready_. Ready means it can be picked up by workers for execution. When the job finishes executing (be it successfully or unsuccessfully, resulting in a failed execution), we signal the semaphore and try to unblock the next job with the same key, if any. Unblocking the next job doesn't mean running that job right away, but moving it from _blocked_ to _ready_. If you're using the `discard` behaviour for `on_conflict`, jobs enqueued while the semaphore is closed will be discarded. Since something can happen that prevents the first job from releasing the semaphore and unblocking the next job (for example, someone pulling a plug in the machine where the worker is running), we have the `duration` as a failsafe. Jobs that have been blocked for more than duration are candidates to be released, but only as many of them as the concurrency rules allow, as each one would need to go through the semaphore dance check. This means that the `duration` is not really about the job that's enqueued or being run, it's about the jobs that are blocked waiting. It's important to note that after one or more candidate jobs are unblocked (either because a job finishes or because `duration` expires and a semaphore is released), the `duration` timer for the still blocked jobs is reset. This happens indirectly via the expiration time of the semaphore, which is updated. In a similar way, when using `discard` as the behaviour to handle conflicts, you might have jobs discarded for up to the `duration` interval if something happens and a running job fails to release the semaphore.
 
 
 For example:
@@ -485,31 +485,6 @@ Jobs are unblocked in order of priority but queue order is not taken into accoun
 
 Finally, failed jobs that are automatically or manually retried work in the same way as new jobs that get enqueued: they get in the queue for getting an open semaphore, and whenever they get it, they'll be run. It doesn't matter if they had already gotten an open semaphore in the past.
 
-### Discarding conflicting jobs
-
-When configuring `on_conflict` with `:discard`, jobs enqueued above the concurrent execution limit are discarded and failed to be enqueued.
-
-```ruby
-class ConcurrentJob < ApplicationJob
-  limits_concurrency key: ->(record) { record }, on_conflict: :discard
-
-  def perform(user); end
-end
-
-enqueued_job = ConcurrentJob.perform_later(record)
-# => instance of ConcurrentJob
-enqueued_job.successfully_enqueued?
-# => true
-
-second_enqueued_job = ConcurrentJob.perform_later(record) do |job|
-  job.successfully_enqueued?
-  # => false
-end
-
-second_enqueued_job
-# => false
-```
-
 ### Performance considerations
 
 Concurrency controls introduce significant overhead (blocked executions need to be created and promoted to ready, semaphores need to be created and updated) so you should consider carefully whether you need them. For throttling purposes, where you plan to have `limit` significantly larger than 1, I'd encourage relying on a limited number of workers per queue instead. For example:
@@ -533,9 +508,8 @@ production:
 
 Or something similar to that depending on your setup. You can also assign a different queue to a job on the moment of enqueuing so you can decide whether to enqueue a job in the throttled queue or another queue depending on the arguments, or pass a block to `queue_as` as explained [here](https://guides.rubyonrails.org/active_job_basics.html#queues).
 
-### Discarding concurrent jobs
-
 
+In addition, mixing concurrency controls with bulk enqueuing (Active Job's `perform_all_later`) is not a good idea because concurrency controlled job needs to be enqueued one by one to ensure concurrency limits are respected, so you lose all the benefits of bulk enqueuing.
 
 ## Failed jobs and retries
 

app/models/solid_queue/job.rb

@@ -2,7 +2,7 @@
 
 module SolidQueue
   class Job < Record
-    class EnqueueError < ActiveJob::EnqueueError; end
+    class EnqueueError < StandardError; end
 
     include Executable, Clearable, Recurrable
 

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, :concurrency_on_conflict, to: :job_class
+        delegate :concurrency_limit, :concurrency_duration, to: :job_class
 
         before_destroy :unblock_next_blocked_job, if: -> { concurrency_limited? && ready? }
       end
@@ -34,8 +34,8 @@ def blocked?
       end
 
       private
-        def discard_concurrent?
-          concurrency_on_conflict == :discard
+        def concurrency_on_conflict
+          job_class.concurrency_on_conflict.to_s.inquiry
         end
 
         def acquire_concurrency_lock
@@ -50,6 +50,14 @@ def release_concurrency_lock
           Semaphore.signal(self)
         end
 
+        def handle_concurrency_conflict
+          if concurrency_on_conflict.discard?
+            finished!
+          else
+            block
+          end
+        end
+
         def block
           BlockedExecution.create_or_find_by!(job_id: id)
         end

app/models/solid_queue/job/executable.rb

@@ -74,16 +74,8 @@ def prepare_for_execution
 
       def dispatch
         if acquire_concurrency_lock then ready
-        elsif discard_concurrent?
-          discard
-          raise EnqueueError.new("Dispatched job discarded due to concurrent configuration.")
         else
-          case job_class.concurrency_on_conflict
-          when :discard
-            discard_on_conflict
-          else
-            block
-          end
+          handle_concurrency_conflict
         end
       end
 
@@ -120,10 +112,6 @@ def ready
           ReadyExecution.create_or_find_by!(job_id: id)
         end
 
-        def discard_on_conflict
-          finished!
-        end
-
         def execution
           %w[ ready claimed failed ].reduce(nil) { |acc, status| acc || public_send("#{status}_execution") }
         end

lib/active_job/concurrency_controls.rb

@@ -5,13 +5,13 @@ module ConcurrencyControls
     extend ActiveSupport::Concern
 
     DEFAULT_CONCURRENCY_GROUP = ->(*) { self.class.name }
+    CONCURRENCY_ON_CONFLICT_BEHAVIOUR = %i[ block discard ]
 
     included do
       class_attribute :concurrency_key, instance_accessor: false
       class_attribute :concurrency_group, default: DEFAULT_CONCURRENCY_GROUP, instance_accessor: false
 
       class_attribute :concurrency_limit
-      class_attribute :concurrency_on_conflict
       class_attribute :concurrency_duration, default: SolidQueue.default_concurrency_control_period
       class_attribute :concurrency_on_conflict, default: :block
     end
@@ -22,7 +22,7 @@ def limits_concurrency(key:, to: 1, group: DEFAULT_CONCURRENCY_GROUP, duration:
         self.concurrency_limit = to
         self.concurrency_group = group
         self.concurrency_duration = duration
-        self.concurrency_on_conflict = on_conflict
+        self.concurrency_on_conflict = on_conflict.presence_in(CONCURRENCY_ON_CONFLICT_BEHAVIOUR) || :block
       end
     end