How can I inspect a queue in ActiveJobs?

Question

Before enqueueing a job I would like to inspect the queue and see if a job already exists in the queue with the exact same arguments and in that case not enqueue the job. But I cannot find out how I should be able to do this. Is it possible?

I know I can easily do it in my tests with the use of the TestHelper. TestHelper relies on the TestAdapter which we of course does not use in the production environment.

A little bit more background. In our API we retrieve the clients version number in each request. We use Intercom for support and want to present the app version in Intercom so we can see what version our customers use when addressing support issues. But to limit the number of calls to Intercom I delay each posting to Intercom a couple of minutes and while a post is enqueued I do not want to enqueue new ones with the same data.

My question is related to List queued tasks with ActiveJob AsyncAdapter but that question only deals with the number of enqueued jobs.

Efficiently reschedule ActiveJob (resque/sidekiq) indicates that this is not possible and I would need to implement the solution separately.

Can I somehow inspect a queue and the jobs in it with ActiveJobs or do I need to keep track of what I have queued and what has been performed?

Answer 1

The following is an unfinished implementation. It only supports Sidekiq at the moment (I assumed you're using Sidekiq).

Note: Have only partially tested this, so you might need to update some things below. I'll check this fully later if I get the time.

class ActiveJob::Base
  def self.where(jid: nil, arguments: [])
    found_jobs = []

    job_adapter = Rails.application.config.active_job.queue_adapter

    case job_adapter

    when :sidekiq
      queue = Sidekiq::Queue.new(queue_name)

      if jid.present?
        job = queue.find_job(jid)

        if job
          # if arguments supplied, check also if args match
          if arguments.present?
            should_disregard_job_class = self == ActiveJob::Base
            job_has_same_class = self.to_s == job.args[0]['job_class']
            job_has_same_arguments = arguments == job.args[0]['arguments'][3..-1]

            if (should_disregard_job_class || job_has_same_class) && job_has_same_arguments
              found_jobs << job
            end

          else
            found_jobs << job
          end
        end

      else
        # TODO: optimise below as it is slow
        queue.each do |job|
          should_disregard_job_class = self == ActiveJob::Base
          job_has_same_class = self.to_s == job.args[0]['job_class']
          job_has_same_arguments = arguments == job.args[0]['arguments'][3..-1]

          if (should_disregard_job_class || job_has_same_class) && job_has_same_arguments
            found_jobs << job
          end 
        end
      end

    when :resque
      # TODO
    else
      raise "TODO: missing Adapter implementation for #{job_adapter}"
    end

    found_jobs
  end
end

Usage

jobs = ActiveJob::Base.where(jid: '12345678abcdef')
jobs = MyCustomJob.where(jid: '12345678abcdef')
jobs = MyCustomJob.where(arguments: ['firstjobargumentvalue', 'secondjobargumentvalue'])