-
-
Save landovsky/8c505ecab41eb38fa1c2cd23058a6ae3 to your computer and use it in GitHub Desktop.
class AddFieldsToDelayedJobs < ActiveRecord::Migration | |
def change | |
add_column :delayed_jobs, :signature, :string, index: true | |
add_column :delayed_jobs, :args, :text | |
end | |
end |
# /lib/delayed_duplicate_prevention_plugin.rb | |
require 'delayed_job' | |
class DelayedDuplicatePreventionPlugin < Delayed::Plugin | |
module SignatureConcern | |
extend ActiveSupport::Concern | |
included do | |
before_validation :add_signature | |
validate :manage_duplicate | |
end | |
attr_writer :strategy | |
@@strategies = [:delete_previous_duplicate, :prevent_duplicate, :allow_duplicate] | |
private | |
def add_signature | |
self.signature = generate_signature | |
self.args = self.payload_object.args.to_yaml | |
end | |
def generate_signature | |
pobj = payload_object | |
if pobj.object.respond_to?(:id) and pobj.object.id.present? | |
sig = "#{pobj.object.class}" | |
sig += ":#{pobj.object.id}" | |
else | |
sig = "#{pobj.object}" | |
end | |
sig += "##{pobj.method_name}" | |
return sig | |
end | |
def manage_duplicate | |
return if @strategy == :allow_duplicate | |
@strategy ||= :prevent_duplicate # default strategy | |
raise "Only the following strategies are permitted: #{@@strategies}" unless @@strategies.include?(@strategy) | |
@checker = DuplicateChecker.new(self) | |
return unless @checker.duplicate? | |
send(@strategy) | |
end | |
def prevent_duplicate | |
if @checker.duplicate? | |
Rails.logger.warn "Found duplicate job(#{self.signature}), ignoring..." | |
errors.add(:base, "This is a duplicate") | |
end | |
end | |
def allow_duplicate | |
end | |
def delete_previous_duplicate | |
if @checker.duplicate? | |
@checker.duplicates.each { |job| Delayed::Job.destroy(job.id) } | |
logger.warn "__________: #{self.class} #{__method__} DELETED #{@checker.duplicates.count} duplicate jobs" | |
end | |
end | |
end | |
class DuplicateChecker | |
attr_reader :job | |
attr_accessor :duplicates | |
def self.duplicate?(job) | |
new(job).duplicate? | |
end | |
def initialize(job) | |
@job = job | |
@duplicates = [] | |
enumerate_duplicates | |
end | |
def enumerate_duplicates | |
possible_dupes = Delayed::Job.where(signature: job.signature) | |
possible_dupes = possible_dupes.where.not(id: job.id) if job.id.present? | |
result = possible_dupes.each do |possible_dupe| | |
duplicates << possible_dupe if possible_dupe.args == job.args | |
end | |
end | |
def duplicate? | |
!duplicates.empty? | |
end | |
private | |
def args_match?(job1, job2) | |
# TODO: make this logic robust | |
normalize_args(job1.args) == normalize_args(job2.args) | |
end | |
def normalize_args(args) | |
args.kind_of?(String) ? YAML.load(args) : args | |
end | |
end | |
end |
# config/initializers/delayed_job.rb | |
require 'delayed_duplicate_prevention_plugin' | |
Delayed::Backend::ActiveRecord::Job.send(:include, DelayedDuplicatePreventionPlugin::SignatureConcern) | |
Delayed::Worker.plugins << DelayedDuplicatePreventionPlugin | |
Here's my try:
https://gist.github.com/vfonic/fe509d53929dfc70d0463c00feec2606
I've removed the strategies, as I only need one strategy for now.
I don't generate MD5 checksum, as I don't see any added value. There's also risk of two MD5 checksums ending up being the same even if the original signature is not the same. It's rare, but why add this layer of complexity.
I've fixed the payload_object.object
by extending the DelayedJob ActiveJob adapter class.
@vfonic Agreed. Thanks for the update. Updated my gist based on yours as I need the same functionality with locked / failed jobs 👍 Simplified it some more using exists?
instead of blank?
for checking existing jobs.
Thought about using jsonb (Postgresql) for the handler column so we could dump the signature column entirely. I don't have the time to investigate how much delayed job is dependent on the YAML format and what would need to change to make that happen. There is an open issue since 2016 (collectiveidea/delayed_job#957) and maybe someone could make that happen :)
Taking some concept from here I implemented a unique job using an active job callback https://gist.github.com/channainfo/b920eeda6b20576310c1fae9780dbedc
I'm not sure why, but I get a no method error for
payload_object.object
andpayload_object.arguments
as mypayload_object
s only have ajob_data
reader method. I'm using Rails, so that might be due to ActiveJob.Anyway, is there any reason for comparing the signature + the arguments? I've revised the code above and simplified it by creating and checking an MD5 checksum using the Job class and arguments: https://gist.github.com/danielpuglisi/08bc1c530064fc03463ed109ef95f598
Would love to hear some feedback.