Feature: Added centralized ActiveJob locking based on lock key. This prevents unnecessary executing/queuing of unique jobs multiple times.

app/jobs/application_job.rb

@@ -10,6 +10,10 @@ class ApplicationJob < ActiveJob::Base
   # This is a workaround to sync ActiveJob#executions to Delayed::Job#attempts
   # until we resolve this dependency.
   after_enqueue do |job|
+    # skip update of `attempts` attribute if job wasn't queued because of ActiveJobLock
+    #(another job with same lock key got queued before this job could be retried)
+    next if job.provider_job_id.blank?
+
     # update the column right away without loading Delayed::Job record
     # see: https://stackoverflow.com/a/34264580
     Delayed::Job.where(id: job.provider_job_id).update_all(attempts: job.executions) # rubocop:disable Rails/SkipsModelValidations

app/jobs/checks_kb_client_notification_job.rb

@@ -1,4 +1,11 @@
 class ChecksKbClientNotificationJob < ApplicationJob
+  include HasActiveJobLock
+
+  def lock_key
+    # "ChecksKbClientNotificationJob/KnowledgeBase::Answer/42/destroy"
+    "#{self.class.name}/#{arguments[0]}/#{arguments[1]}/#{arguments[2]}"
+  end
+
   def perform(klass_name, id, event)
     object = klass_name.constantize.find_by(id: id)
     return if object.blank?

app/jobs/concerns/has_active_job_lock.rb

@@ -0,0 +1,118 @@
+module HasActiveJobLock
+  extend ActiveSupport::Concern
+
+  included do
+    before_enqueue do |job| # rubocop:disable Style/SymbolProc
+      job.ensure_active_job_lock_for_enqueue!
+    end
+
+    around_perform do |job, block|
+      job.mark_active_job_lock_as_started
+
+      block.call
+    ensure
+      job.release_active_job_lock!
+    end
+  end
+
+  # Defines the lock key for the current job to prevent execution of jobs with the same key.
+  # This is by default the name of the ActiveJob class.
+  # If you're in the situation where you need to have a lock_key based on
+  # the given arguments you can overwrite this method in your job and access
+  # them via `arguments`. See ActiveJob::Core for more (e.g. queue).
+  #
+  # @example
+  #  # default
+  #  job = UniqueActiveJob.new
+  #  job.lock_key
+  #  # => "UniqueActiveJob"
+  #
+  # @example
+  #  # with lock_key: "#{self.class.name}/#{arguments[0]}/#{arguments[1]}"
+  #  job = SearchIndexJob.new('User', 42)
+  #  job.lock_key
+  #  # => "SearchIndexJob/User/42"
+  #
+  # return [String]
+  def lock_key
+    self.class.name
+  end
+
+  def mark_active_job_lock_as_started
+    release_active_job_lock_cache
+
+    in_active_job_lock_transaction do
+      # a perform_now job doesn't require any locking
+      return if active_job_lock.blank?
+      return if !active_job_lock.of?(self)
+
+      # a perform_later job started to perform and will be marked as such
+      active_job_lock.touch # rubocop:disable Rails/SkipsModelValidations
+    end
+  end
+
+  def ensure_active_job_lock_for_enqueue!
+    release_active_job_lock_cache
+
+    in_active_job_lock_transaction do
+      return if active_job_lock_for_enqueue!.present?
+
+      ActiveJobLock.create!(
+        lock_key:      lock_key,
+        active_job_id: job_id,
+      )
+    end
+  end
+
+  def release_active_job_lock!
+    # only delete lock if the current job is the one holding the lock
+    # perform_now jobs or perform_later jobs for which follow-up jobs were enqueued
+    # don't need to remove any locks
+    lock = ActiveJobLock.lock.find_by(lock_key: lock_key, active_job_id: job_id)
+
+    if !lock
+      logger.debug { "Found no ActiveJobLock for #{self.class.name} (Job ID: #{job_id}) with key '#{lock_key}'." }
+      return
+    end
+
+    logger.debug { "Deleting ActiveJobLock for #{self.class.name} (Job ID: #{job_id}) with key '#{lock_key}'." }
+    lock.destroy!
+  end
+
+  private
+
+  def in_active_job_lock_transaction
+    # re-use active DB transaction if present
+    return yield if ActiveRecord::Base.connection.open_transactions.nonzero?
+
+    # start own serializable DB transaction to prevent race conditions on DB level
+    ActiveJobLock.transaction(isolation: :serializable) do
+      yield
+    end
+  rescue ActiveRecord::RecordNotUnique
+    existing_active_job_lock!
+  end
+
+  def active_job_lock_for_enqueue!
+    return if active_job_lock.blank?
+
+    # don't enqueue perform_later jobs if a job with the same
+    # lock key exists that hasn't started to perform yet
+    existing_active_job_lock! if active_job_lock.peform_pending?
+
+    active_job_lock.tap { |lock| lock.transfer_to(self) }
+  end
+
+  def active_job_lock
+    @active_job_lock ||= ActiveJobLock.lock.find_by(lock_key: lock_key)
+  end
+
+  def release_active_job_lock_cache
+    @active_job_lock = nil
+  end
+
+  def existing_active_job_lock!
+    logger.info "Won't enqueue #{self.class.name} (Job ID: #{job_id}) because of already existing job with lock key '#{lock_key}'."
+    throw :abort
+  end
+end

app/jobs/scheduled_touch_job.rb

@@ -1,9 +1,16 @@
 class ScheduledTouchJob < ApplicationJob
-  def perform(klass_name, id)
-    klass_name.constantize.find_by(id: id)&.touch # rubocop:disable Rails/SkipsModelValidations
+  include HasActiveJobLock
+
+  def lock_key
+    # "ScheduledTouchJob/User/42"
+    "#{self.class.name}/#{arguments[0]}/#{arguments[1]}"
   end
 
   def self.touch_at(object, date)
     set(wait_until: date).perform_later(object.class.to_s, object.id)
   end
+
+  def perform(klass_name, id)
+    klass_name.constantize.find_by(id: id)&.touch # rubocop:disable Rails/SkipsModelValidations
+  end
 end

app/jobs/search_index_job.rb

@@ -1,9 +1,15 @@
 class SearchIndexJob < ApplicationJob
+  include HasActiveJobLock
 
   retry_on StandardError, attempts: 20, wait: lambda { |executions|
     executions * 10.seconds
   }
 
+  def lock_key
+    # "SearchIndexJob/User/42"
+    "#{self.class.name}/#{arguments[0]}/#{arguments[1]}"
+  end
+
   def perform(object, o_id)
     @object = object
     @o_id   = o_id

app/jobs/sla_ticket_rebuild_escalation_job.rb

@@ -1,4 +1,6 @@
 class SlaTicketRebuildEscalationJob < ApplicationJob
+  include HasActiveJobLock
+
   def perform
     Cache.delete('SLA::List::Active')
     Ticket::Escalation.rebuild_all

app/jobs/ticket_online_notification_seen_job.rb

@@ -1,4 +1,11 @@
 class TicketOnlineNotificationSeenJob < ApplicationJob
+  include HasActiveJobLock
+
+  def lock_key
+    # "TicketOnlineNotificationSeenJob/23/42"
+    "#{self.class.name}/#{arguments[0]}/#{arguments[1]}"
+  end
+
   def perform(ticket_id, user_id)
     user_id = user_id || 1
 

app/jobs/ticket_user_ticket_counter_job.rb

@@ -1,4 +1,11 @@
 class TicketUserTicketCounterJob < ApplicationJob
+  include HasActiveJobLock
+
+  def lock_key
+    # "TicketUserTicketCounterJob/23/42"
+    "#{self.class.name}/#{arguments[0]}/#{arguments[1]}"
+  end
+
   def perform(customer_id, updated_by_id)
 
     # check if update is needed

app/models/active_job_lock.rb

@@ -0,0 +1,21 @@
+class ActiveJobLock < ApplicationModel
+
+  def of?(active_job)
+    active_job.job_id == active_job_id
+  end
+
+  def peform_pending?
+    updated_at == created_at
+  end
+
+  def transfer_to(active_job)
+    logger.info "Transferring ActiveJobLock with id '#{id}' from active_job_id '#{active_job_id}' to active_job_id '#{active_job_id}'."
+
+    reset_time_stamp = Time.zone.now
+    update!(
+      active_job_id: active_job.job_id,
+      created_at:    reset_time_stamp,
+      updated_at:    reset_time_stamp
+    )
+  end
+end

db/migrate/20120101000001_create_base.rb

@@ -715,5 +715,14 @@ class CreateBase < ActiveRecord::Migration[4.2]
     add_index :http_logs, [:created_at]
     add_foreign_key :http_logs, :users, column: :created_by_id
     add_foreign_key :http_logs, :users, column: :updated_by_id
+
+    create_table :active_job_locks do |t|
+      t.string :lock_key
+      t.string :active_job_id
+
+      t.timestamps
+    end
+    add_index :active_job_locks, :lock_key, unique: true
+    add_index :active_job_locks, :active_job_id, unique: true
   end
 end