Sidekiq throttled queueing

20180710014306_create_resource_throttles.rb

class CreateResourceThrottles < ActiveRecord::Migration[5.1]
  def change
    create_table :resource_throttles do |t|
      t.string :throttle_key, null: false
      t.string :resource_uid, null: false
      t.boolean :active, null: false, default: false

      t.index [:throttle_key, :resource_uid], unique: true

      t.timestamps
    end
  end
end

enqueue_throttler.rb

# sidekiq-throttled gem controls only concurrent execution
# and allow to enqueue many jobs in queue which can cause queues growth
# This throttler doesn't allow to put job in queue if the same job already in queue or processing
module EnqueueThrottler
  extend ActiveSupport::Concern

  module ClassMethods
    def perform_async(resource_uid, *args)
      if ResourceThrottle.capture!(self.name, resource_uid)
        super
      end
    end
  end

  def self.prepended(base)
    class << base
      prepend ClassMethods
    end
  end

  def perform(resource_uid, *args)
    super
  ensure
    ResourceThrottle.release!(self.class.name, resource_uid)
  end
end

resource_throttle.rb

class ResourceThrottle < ApplicationRecord
  validates :throttle_key, presence: true
  validates :resource_uid, presence: true
  validates :resource_uid, uniqueness: { scope: :throttle_key }

  class << self
    def capture!(throttle_key, resource_uid)
      throttle = self.find_or_create_by!(throttle_key: throttle_key, resource_uid: resource_uid)

      if !throttle.active
        throttle.update!(active: true)
        true
      end
    end

    def release!(throttle_key, resource_uid)
      throttle = self.find_or_create_by!(throttle_key: throttle_key, resource_uid: resource_uid)
      throttle.update!(active: false)
    end
  end
end

sample_worker.rb

class SampleWorker
  include Sidekiq::Worker
  include Sidekiq::Throttled::Worker
  prepend EnqueueThrottler


  sidekiq_options retry: false, queue: :low

  # Allow maximum 1 concurrent job per project at a time. uses gem `sidekiq-throttled`
  sidekiq_throttle concurrency: {
    limit: 1,
    key_suffix: ->(project_id) { project_id }
  }

  def perform(project_id)
    # perform heavy job
  end
end