Fix cursor assignment

.reek.yml

@@ -36,7 +36,7 @@ detectors:
       - SidekiqUniqueJobs::Locksmith#set_lock_info
       - SidekiqUniqueJobs::Locksmith#wait_for_primed_token
       - SidekiqUniqueJobs::OnConflict::Reschedule#call
-      - SidekiqUniqueJobs::Orphans::Reaper#entries
+      - SidekiqUniqueJobs::Orphans::RubyReaper#entries
       - SidekiqUniqueJobs::Profiler#self.stop
       - SidekiqUniqueJobs::Script::Caller#extract_args
       - SidekiqUniqueJobs::Timing#timed
@@ -51,8 +51,8 @@ detectors:
       - SidekiqUniqueJobs::Logging#debug_item
       - SidekiqUniqueJobs::NotUniqueWorker#initialize
       - SidekiqUniqueJobs::OnConflict::Reject#push_to_deadset
-      - SidekiqUniqueJobs::Orphans::Reaper#active?
-      - SidekiqUniqueJobs::Orphans::Reaper#entries
+      - SidekiqUniqueJobs::Orphans::RubyReaper#active?
+      - SidekiqUniqueJobs::Orphans::RubyReaper#entries
       - SidekiqUniqueJobs::SidekiqWorkerMethods#worker_class_constantize
       - SidekiqUniqueJobs::Web::Helpers#cparams
   IrresponsibleModule:
@@ -91,8 +91,8 @@ detectors:
       - SidekiqUniqueJobs::Locksmith#create_lock
       - SidekiqUniqueJobs::Middleware#self.configure_client
       - SidekiqUniqueJobs::Middleware#self.configure_server
-      - SidekiqUniqueJobs::Orphans::Reaper#active?
-      - SidekiqUniqueJobs::Orphans::Reaper#enqueued?
+      - SidekiqUniqueJobs::Orphans::RubyReaper#active?
+      - SidekiqUniqueJobs::Orphans::RubyReaper#enqueued?
       - SidekiqUniqueJobs::UpgradeLocks#keys_for_digest
   RepeatedConditional:
     exclude:
@@ -126,9 +126,9 @@ detectors:
       - SidekiqUniqueJobs::Middleware#call
       - SidekiqUniqueJobs::Middleware#self.configure_server
       - SidekiqUniqueJobs::Orphans::Manager#start
-      - SidekiqUniqueJobs::Orphans::Reaper#active?
-      - SidekiqUniqueJobs::Orphans::Reaper#enqueued?
-      - SidekiqUniqueJobs::Orphans::Reaper#entries
+      - SidekiqUniqueJobs::Orphans::RubyReaper#active?
+      - SidekiqUniqueJobs::Orphans::RubyReaper#enqueued?
+      - SidekiqUniqueJobs::Orphans::RubyReaper#entries
       - SidekiqUniqueJobs::Profiler#self.stop
       - SidekiqUniqueJobs::Script::Caller#call_script
       - SidekiqUniqueJobs::Script::Caller#extract_args

lib/sidekiq_unique_jobs/constants.rb

@@ -30,6 +30,7 @@ module SidekiqUniqueJobs
   ON_CLIENT_CONFLICT    ||= "on_client_conflict"
   ON_CONFLICT           ||= "on_conflict"
   ON_SERVER_CONFLICT    ||= "on_server_conflict"
+  PROCESSES             ||= "processes"
   QUEUE                 ||= "queue"
   RETRY                 ||= "retry"
   SCHEDULE              ||= "schedule"

lib/sidekiq_unique_jobs/lua/shared/_find_digest_in_queues.lua

@@ -34,12 +34,9 @@ local function find_digest_in_queues(digest)
         end
         index = index + per
       end
-
-      cursor = next_cursor
-      if cursor == "0" then
-        log_debug("Looped through all queues, stopping iteration")
-      end
     end
+
+    cursor = next_cursor
   until found == true or cursor == "0"
 
   return result

lib/sidekiq_unique_jobs/orphans/lua_reaper.rb

@@ -0,0 +1,29 @@
+# frozen_string_literal: true
+
+module SidekiqUniqueJobs
+  module Orphans
+    #
+    # Class DeleteOrphans provides deletion of orphaned digests
+    #
+    # @note this is a much slower version of the lua script but does not crash redis
+    #
+    # @author Mikael Henriksson <mikael@zoolutions.se>
+    #
+    class LuaReaper < Reaper
+      #
+      # Delete orphaned digests
+      #
+      #
+      # @return [Integer] the number of reaped locks
+      #
+      def call
+        call_script(
+          :reap_orphans,
+          conn,
+          keys: [DIGESTS, SCHEDULE, RETRY, PROCESSES],
+          argv: [reaper_count],
+        )
+      end
+    end
+  end
+end

lib/sidekiq_unique_jobs/orphans/reaper.rb

@@ -9,13 +9,20 @@ module Orphans
     #
     # @author Mikael Henriksson <mikael@zoolutions.se>
     #
-    # rubocop:disable Metrics/ClassLength
     class Reaper
       include SidekiqUniqueJobs::Connection
       include SidekiqUniqueJobs::Script::Caller
       include SidekiqUniqueJobs::Logging
       include SidekiqUniqueJobs::JSON
 
+      require_relative "lua_reaper"
+      require_relative "ruby_reaper"
+
+      REAPERS = {
+        lua: SidekiqUniqueJobs::Orphans::LuaReaper,
+        ruby: SidekiqUniqueJobs::Orphans::RubyReaper,
+      }.freeze
+
       #
       # Execute deletion of orphaned digests
       #
@@ -29,18 +36,18 @@ def self.call(conn = nil)
         redis { |rcon| new(rcon).call }
       end
 
-      attr_reader :conn, :digests, :scheduled, :retried
+      #
+      # @!attribute [r] conn
+      #   @return [Redis] a redis connection
+      attr_reader :conn
 
       #
       # Initialize a new instance of DeleteOrphans
       #
       # @param [Redis] conn a connection to redis
       #
       def initialize(conn)
-        @conn      = conn
-        @digests   = SidekiqUniqueJobs::Digests.new
-        @scheduled = Redis::SortedSet.new(SCHEDULE)
-        @retried   = Redis::SortedSet.new(RETRY)
+        @conn = conn
       end
 
       #
@@ -80,175 +87,12 @@ def reaper_count
       # @return [Integer] the number of reaped locks
       #
       def call
-        case reaper
-        when :ruby
-          execute_ruby_reaper
-        when :lua
-          execute_lua_reaper
+        if (implementation = REAPERS[reaper])
+          implementation.new(conn).call
         else
           log_fatal(":#{reaper} is invalid for `SidekiqUnqiueJobs.config.reaper`")
         end
       end
-
-      #
-      # Executes the ruby reaper
-      #
-      #
-      # @return [Integer] the number of deleted locks
-      #
-      def execute_ruby_reaper
-        BatchDelete.call(orphans, conn)
-      end
-
-      #
-      # Executes the lua reaper
-      #
-      #
-      # @return [Integer] the number of deleted locks
-      #
-      def execute_lua_reaper
-        call_script(
-          :reap_orphans,
-          conn,
-          keys: [SidekiqUniqueJobs::DIGESTS, SidekiqUniqueJobs::SCHEDULE, SidekiqUniqueJobs::RETRY],
-          argv: [reaper_count],
-        )
-      end
-
-      #
-      # Find orphaned digests
-      #
-      #
-      # @return [Array<String>] an array of orphaned digests
-      #
-      def orphans
-        conn.zrevrange(digests.key, 0, -1).each_with_object([]) do |digest, result|
-          next if belongs_to_job?(digest)
-
-          result << digest
-          break if result.size >= reaper_count
-        end
-      end
-
-      #
-      # Checks if the digest has a matching job.
-      #   1. It checks the scheduled set
-      #   2. It checks the retry set
-      #   3. It goes through all queues
-      #
-      #
-      # @param [String] digest the digest to search for
-      #
-      # @return [true] when either of the checks return true
-      # @return [false] when no job was found for this digest
-      #
-      def belongs_to_job?(digest)
-        scheduled?(digest) || retried?(digest) || enqueued?(digest) || active?(digest)
-      end
-
-      #
-      # Checks if the digest exists in the Sidekiq::ScheduledSet
-      #
-      # @param [String] digest the current digest
-      #
-      # @return [true] when digest exists in scheduled set
-      #
-      def scheduled?(digest)
-        in_sorted_set?(SCHEDULE, digest)
-      end
-
-      #
-      # Checks if the digest exists in the Sidekiq::RetrySet
-      #
-      # @param [String] digest the current digest
-      #
-      # @return [true] when digest exists in retry set
-      #
-      def retried?(digest)
-        in_sorted_set?(RETRY, digest)
-      end
-
-      #
-      # Checks if the digest exists in a Sidekiq::Queue
-      #
-      # @param [String] digest the current digest
-      #
-      # @return [true] when digest exists in any queue
-      #
-      def enqueued?(digest)
-        Sidekiq.redis do |conn|
-          queues(conn) do |queue|
-            entries(conn, queue) do |entry|
-              return true if entry.include?(digest)
-            end
-          end
-
-          false
-        end
-      end
-
-      def active?(digest)
-        Sidekiq.redis do |conn|
-          procs = conn.sscan_each("processes").to_a.sort
-
-          result = conn.pipelined do
-            procs.map do |key|
-              conn.hget(key, "info")
-            end
-          end
-
-          result.flatten.compact.any? { |job| load_json(job)[LOCK_DIGEST] == digest }
-        end
-      end
-
-      #
-      # Loops through all the redis queues and yields them one by one
-      #
-      # @param [Redis] conn the connection to use for fetching queues
-      #
-      # @return [void]
-      #
-      # @yield queues one at a time
-      #
-      def queues(conn, &block)
-        conn.sscan_each("queues", &block)
-      end
-
-      def entries(conn, queue) # rubocop:disable Metrics/MethodLength
-        queue_key    = "queue:#{queue}"
-        initial_size = conn.llen(queue_key)
-        deleted_size = 0
-        page         = 0
-        page_size    = 50
-
-        loop do
-          range_start = page * page_size - deleted_size
-          range_end   = range_start + page_size - 1
-          entries     = conn.lrange(queue_key, range_start, range_end)
-          page       += 1
-
-          entries.each do |entry|
-            yield entry
-          end
-
-          deleted_size = initial_size - conn.llen(queue_key)
-        end
-      end
-
-      #
-      # Checks a sorted set for the existance of this digest
-      #
-      #
-      # @param [String] key the key for the sorted set
-      # @param [String] digest the digest to scan for
-      #
-      # @return [true] when found
-      # @return [false] when missing
-      #
-      def in_sorted_set?(key, digest)
-        conn.zscan_each(key, match: "*#{digest}*", count: 1).to_a.any?
-      end
     end
-    # rubocop:enable Metrics/ClassLength
   end
 end