diff options
Diffstat (limited to 'lib/gitlab/sidekiq_middleware/duplicate_jobs')
6 files changed, 110 insertions, 42 deletions
diff --git a/lib/gitlab/sidekiq_middleware/duplicate_jobs/strategies.rb b/lib/gitlab/sidekiq_middleware/duplicate_jobs/strategies.rb index 6fdef4c354e..63e8bee4443 100644 --- a/lib/gitlab/sidekiq_middleware/duplicate_jobs/strategies.rb +++ b/lib/gitlab/sidekiq_middleware/duplicate_jobs/strategies.rb @@ -8,6 +8,7 @@ module Gitlab STRATEGIES = { until_executing: UntilExecuting, + until_executed: UntilExecuted, none: None }.freeze diff --git a/lib/gitlab/sidekiq_middleware/duplicate_jobs/strategies/base.rb b/lib/gitlab/sidekiq_middleware/duplicate_jobs/strategies/base.rb new file mode 100644 index 00000000000..df5df590281 --- /dev/null +++ b/lib/gitlab/sidekiq_middleware/duplicate_jobs/strategies/base.rb @@ -0,0 +1,37 @@ +# frozen_string_literal: true + +module Gitlab + module SidekiqMiddleware + module DuplicateJobs + module Strategies + class Base + def initialize(duplicate_job) + @duplicate_job = duplicate_job + end + + def schedule(job) + raise NotImplementedError + end + + def perform(_job) + raise NotImplementedError + end + + private + + attr_reader :duplicate_job + + def strategy_name + self.class.name.to_s.demodulize.underscore.humanize.downcase + end + + def check! + # The default expiry time is the DuplicateJob::DUPLICATE_KEY_TTL already + # Only the strategies de-duplicating when scheduling + duplicate_job.check! + end + end + end + end + end +end diff --git a/lib/gitlab/sidekiq_middleware/duplicate_jobs/strategies/deduplicates_when_scheduling.rb b/lib/gitlab/sidekiq_middleware/duplicate_jobs/strategies/deduplicates_when_scheduling.rb new file mode 100644 index 00000000000..59b0e7e29da --- /dev/null +++ b/lib/gitlab/sidekiq_middleware/duplicate_jobs/strategies/deduplicates_when_scheduling.rb @@ -0,0 +1,47 @@ +# frozen_string_literal: true + +module Gitlab + module SidekiqMiddleware + module DuplicateJobs + module Strategies + module DeduplicatesWhenScheduling + def initialize(duplicate_job) + @duplicate_job = duplicate_job + end + + def schedule(job) + if deduplicatable_job? && check! && duplicate_job.duplicate? + job['duplicate-of'] = duplicate_job.existing_jid + + if duplicate_job.droppable? + Gitlab::SidekiqLogging::DeduplicationLogger.instance.log( + job, "dropped #{strategy_name}", duplicate_job.options) + return false + end + end + + yield + end + + private + + def deduplicatable_job? + !duplicate_job.scheduled? || duplicate_job.options[:including_scheduled] + end + + def check! + duplicate_job.check!(expiry) + end + + def expiry + return DuplicateJob::DUPLICATE_KEY_TTL unless duplicate_job.scheduled? + + time_diff = duplicate_job.scheduled_at.to_i - Time.now.to_i + + time_diff > 0 ? time_diff : DuplicateJob::DUPLICATE_KEY_TTL + end + end + end + end + end +end diff --git a/lib/gitlab/sidekiq_middleware/duplicate_jobs/strategies/none.rb b/lib/gitlab/sidekiq_middleware/duplicate_jobs/strategies/none.rb index cd101cd16b6..acbe0efaafa 100644 --- a/lib/gitlab/sidekiq_middleware/duplicate_jobs/strategies/none.rb +++ b/lib/gitlab/sidekiq_middleware/duplicate_jobs/strategies/none.rb @@ -5,10 +5,7 @@ module Gitlab module DuplicateJobs module Strategies # This strategy will never deduplicate a job - class None - def initialize(_duplicate_job) - end - + class None < Base def schedule(_job) yield end diff --git a/lib/gitlab/sidekiq_middleware/duplicate_jobs/strategies/until_executed.rb b/lib/gitlab/sidekiq_middleware/duplicate_jobs/strategies/until_executed.rb new file mode 100644 index 00000000000..738efa36fc8 --- /dev/null +++ b/lib/gitlab/sidekiq_middleware/duplicate_jobs/strategies/until_executed.rb @@ -0,0 +1,22 @@ +# frozen_string_literal: true + +module Gitlab + module SidekiqMiddleware + module DuplicateJobs + module Strategies + # This strategy takes a lock before scheduling the job in a queue and + # removes the lock after the job has executed preventing a new job to be queued + # while a job is still executing. + class UntilExecuted < Base + include DeduplicatesWhenScheduling + + def perform(_job) + yield + + duplicate_job.delete! + end + end + end + end + end +end diff --git a/lib/gitlab/sidekiq_middleware/duplicate_jobs/strategies/until_executing.rb b/lib/gitlab/sidekiq_middleware/duplicate_jobs/strategies/until_executing.rb index 46ce0eb4a91..68d66383b2b 100644 --- a/lib/gitlab/sidekiq_middleware/duplicate_jobs/strategies/until_executing.rb +++ b/lib/gitlab/sidekiq_middleware/duplicate_jobs/strategies/until_executing.rb @@ -7,50 +7,14 @@ module Gitlab # This strategy takes a lock before scheduling the job in a queue and # removes the lock before the job starts allowing a new job to be queued # while a job is still executing. - class UntilExecuting - def initialize(duplicate_job) - @duplicate_job = duplicate_job - end - - def schedule(job) - if deduplicatable_job? && check! && duplicate_job.duplicate? - job['duplicate-of'] = duplicate_job.existing_jid - - if duplicate_job.droppable? - Gitlab::SidekiqLogging::DeduplicationLogger.instance.log( - job, "dropped until executing", duplicate_job.options) - return false - end - end - - yield - end + class UntilExecuting < Base + include DeduplicatesWhenScheduling def perform(_job) duplicate_job.delete! yield end - - private - - attr_reader :duplicate_job - - def deduplicatable_job? - !duplicate_job.scheduled? || duplicate_job.options[:including_scheduled] - end - - def check! - duplicate_job.check!(expiry) - end - - def expiry - return DuplicateJob::DUPLICATE_KEY_TTL unless duplicate_job.scheduled? - - time_diff = duplicate_job.scheduled_at.to_i - Time.now.to_i - - time_diff > 0 ? time_diff : DuplicateJob::DUPLICATE_KEY_TTL - end end end end |