app/workers/redis_migration_worker.rb


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

# frozen_string_literal: true

class RedisMigrationWorker
  include ApplicationWorker

  idempotent!
  data_consistency :delayed
  feature_category :redis
  urgency :throttled
  loggable_arguments 0

  SCAN_START_STOP = '0'

  def perform(job_class_name, cursor, options = {})
    migrator = self.class.fetch_migrator!(job_class_name)

    scan_size = options[:scan_size] || 1000
    deadline = Time.now.utc + 3.minutes

    while Time.now.utc < deadline
      cursor, keys = migrator.redis.scan(cursor, match: migrator.scan_match_pattern, count: scan_size)

      migrator.perform(keys) if keys.any?

      sleep(0.01)
      break if cursor == SCAN_START_STOP
    end

    self.class.perform_async(job_class_name, cursor, options) unless cursor == SCAN_START_STOP
  end

  class << self
    def fetch_migrator!(job_class_name)
      job_class = "Gitlab::BackgroundMigration::Redis::#{job_class_name}".safe_constantize
      raise NotImplementedError, "#{job_class_name} does not exist" if job_class.nil?

      job_class.new
    end
  end
end