class SidekiqUniqueJobs::UpgradeLocks
Upgrades locks between gem version upgrades
@author Mikael Henriksson <mikael@mhenrixon.com>
Constants
- BATCH_SIZE
-
@return [Integer] the number of keys to batch upgrade
- OLD_SUFFIXES
-
@return [Array<String>] suffixes for old version
Attributes
Public Class Methods
Source
# File lib/sidekiq_unique_jobs/upgrade_locks.rb, line 31 def self.call redis do |conn| new(conn).call end end
Performs upgrade of old locks
@return [Integer] the number of upgrades locks
Source
# File lib/sidekiq_unique_jobs/upgrade_locks.rb, line 39 def initialize(conn) @count = 0 @conn = conn redis_version # Avoid pipelined calling redis_version and getting a future. end
Public Instance Methods
Source
# File lib/sidekiq_unique_jobs/upgrade_locks.rb, line 51 def call with_logging_context do return log_info("Already upgraded to #{version}") if conn.hget(upgraded_key, version) # TODO: Needs handling of v7.0.0 => v7.0.1 where we don't want to return log_info("Skipping upgrade because #{DEAD_VERSION} has been set") if conn.get(DEAD_VERSION) log_info("Start - Upgrading Locks") # upgrade_v6_locks # delete_unused_v6_keys # delete_supporting_v6_keys conn.hset(upgraded_key, version, now_f) log_info("Done - Upgrading Locks") end @count end
Performs upgrade of old locks
@return [Integer] the number of upgrades locks
Private Instance Methods
Source
# File lib/sidekiq_unique_jobs/upgrade_locks.rb, line 115 def batch_delete(*keys) return if keys.empty? conn.pipelined do |pipeline| pipeline.unlink(*keys) end end
Source
# File lib/sidekiq_unique_jobs/upgrade_locks.rb, line 123 def batch_scan(match:, count:) cursor = "0" loop do cursor, values = conn.scan(cursor, match: match, count: count) yield values break if cursor == "0" end end
Source
# File lib/sidekiq_unique_jobs/upgrade_locks.rb, line 109 def delete_suffix(suffix) batch_scan(match: "*:#{suffix}", count: BATCH_SIZE) do |keys| batch_delete(*keys) end end
Source
# File lib/sidekiq_unique_jobs/upgrade_locks.rb, line 105 def delete_supporting_v6_keys batch_delete("unique:keys") end
Source
# File lib/sidekiq_unique_jobs/upgrade_locks.rb, line 97 def delete_unused_v6_keys log_info("Start - Deleting v6 keys") OLD_SUFFIXES.each do |suffix| delete_suffix(suffix) end log_info("Done - Deleting v6 keys") end
Source
# File lib/sidekiq_unique_jobs/upgrade_locks.rb, line 144 def logging_context if logger_context_hash? { "uniquejobs" => :upgrade_locks } else "uniquejobs-upgrade_locks" end end
Source
# File lib/sidekiq_unique_jobs/upgrade_locks.rb, line 136 def now_f SidekiqUniqueJobs.now_f end
Source
# File lib/sidekiq_unique_jobs/upgrade_locks.rb, line 140 def redis_version @redis_version ||= SidekiqUniqueJobs.config.redis_version end
Source
# File lib/sidekiq_unique_jobs/upgrade_locks.rb, line 86 def upgrade_v6_lock(grabbed_key) locked_key = grabbed_key.gsub(":GRABBED", ":LOCKED") digest = grabbed_key.gsub(":GRABBED", "") locks = conn.hgetall(grabbed_key) conn.pipelined do |pipeline| pipeline.hmset(locked_key, *locks.to_a) pipeline.zadd(DIGESTS, locks.values.first, digest) end end
Source
# File lib/sidekiq_unique_jobs/upgrade_locks.rb, line 76 def upgrade_v6_locks log_info("Start - Converting v6 locks to v7") conn.scan(match: "*:GRABBED", count: BATCH_SIZE).each do |grabbed_key| upgrade_v6_lock(grabbed_key) @count += 1 end log_info("Done - Converting v6 locks to v7") end
Source
# File lib/sidekiq_unique_jobs/upgrade_locks.rb, line 72 def upgraded_key @upgraded_key ||= "#{LIVE_VERSION}:UPGRADED" end
Source
# File lib/sidekiq_unique_jobs/upgrade_locks.rb, line 132 def version SidekiqUniqueJobs.version end