Bug fixes

README.md

@@ -361,7 +361,7 @@ For sidekiq versions before 5.1 a `sidekiq_retries_exhausted` block is required
 ```ruby
 class MyWorker
   sidekiq_retries_exhausted do |msg, _ex|
-    SidekiqUniqueJobs::Digests.delete_by(digest: msg['unique_digest']) if msg['unique_digest']
+    SidekiqUniqueJobs::Digests.del(digest: msg['unique_digest']) if msg['unique_digest']
   end
 end
 ```
@@ -372,7 +372,7 @@ Starting in v5.1, Sidekiq can also fire a global callback when a job dies:
 # this goes in your initializer
 Sidekiq.configure_server do |config|
   config.death_handlers << ->(job, _ex) do
-    SidekiqUniqueJobs::Digests.delete_by(digest: job['unique_digest']) if job['unique_digest']
+    SidekiqUniqueJobs::Digests.del(digest: job['unique_digest']) if job['unique_digest']
   end
 end
 ```

lib/sidekiq_unique_jobs/digests.rb

@@ -22,6 +22,23 @@ def all(pattern: SCAN_PATTERN, count: DEFAULT_COUNT)
       redis { |conn| conn.sscan_each(UNIQUE_SET, match: pattern, count: count).to_a }
     end
 
+    # Paginate unique digests
+    #
+    # @param [String] pattern a pattern to match with
+    # @param [Integer] page the current cursor position
+    # @param [Integer] count the maximum number to match
+    # @return [Array<String>] with unique digests
+    def page(pattern: SCAN_PATTERN, cursor: 0, page_size: 100)
+      redis do |conn|
+        total_size, digests = conn.multi do
+          conn.scard(UNIQUE_SET)
+          conn.sscan(UNIQUE_SET, cursor, match: pattern, count: page_size)
+        end
+
+        [total_size, digests[0], digests[1]]
+      end
+    end
+
     # Get a total count of unique digests
     #
     # @return [Integer] number of digests

lib/sidekiq_unique_jobs/locksmith.rb

@@ -32,6 +32,16 @@ def create
         keys: [exists_key, grabbed_key, available_key, version_key, UNIQUE_SET, unique_digest],
         argv: [jid, expiration, API_VERSION, concurrency],
       )
+      expire
+    end
+
+    def expire
+      Scripts.call(
+        :expire,
+        redis_pool,
+        keys: [exists_key, available_key, version_key],
+        argv: [expiration],
+      )
     end
 
     # Checks if the exists key is created in redis
@@ -86,7 +96,7 @@ def unlock
       signal(jid)
     end
 
-    # Removes the lock keys from Redis
+    # Checks if this instance is considered locked
     # @param [String] token the unique token to check for a lock.
     #   nil will default to the jid provided in the initializer
     # @return [true, false]

lib/sidekiq_unique_jobs/web.rb

@@ -20,11 +20,13 @@ def self.registered(app) # rubocop:disable Metrics/MethodLength
       end
 
       app.get '/unique_digests' do
-        @total_size     = Digests.count
         @filter         = params[:filter] || '*'
         @filter         = '*' if @filter == ''
         @count          = (params[:count] || 100).to_i
-        @unique_digests = Digests.all(pattern: @filter, count: @count)
+        @current_cursor = params[:cursor]
+        @prev_cursor    = params[:prev_cursor]
+        @total_size, @next_cursor, @unique_digests =
+          Digests.page(pattern: @filter, cursor: @current_cursor, page_size: @count)
 
         erb(unique_template(:unique_digests))
       end

lib/sidekiq_unique_jobs/web/helpers.rb

@@ -9,6 +9,20 @@ def unique_template(name)
         File.open(File.join(VIEW_PATH, "#{name}.erb")).read
       end
 
+      SAFE_CPARAMS = %w[cursor prev_cursor].freeze
+
+      def cparams(options)
+        # stringify
+        options.keys.each do |key|
+          options[key.to_s] = options.delete(key)
+        end
+
+        params.merge(options).map do |key, value|
+          next unless SAFE_CPARAMS.include?(key)
+          "#{key}=#{CGI.escape(value.to_s)}"
+        end.compact.join('&')
+      end
+
       def redirect_to(subpath)
         if respond_to?(:to)
           # Sinatra-based web UI

lib/sidekiq_unique_jobs/web/views/_paging.erb

@@ -0,0 +1,10 @@
+<ul class="pagination pull-right flip">
+  <% if @prev_cursor %>
+    <li>
+      <a href="<%= url %>?<%= cparams(cursor: @prev_cursor, prev_cursor: @next_cursor) %>">Previous <%= @count %></a>
+    </li>
+  <% end %>
+  <li>
+    <a href="<%= url %>?<%= cparams(cursor: @next_cursor, prev_cursor: @current_cursor) %>">Next <%= @count %></a>
+  </li>
+</ul>

lib/sidekiq_unique_jobs/web/views/unique_digests.erb

@@ -9,7 +9,7 @@
   </form>
   <% if @unique_digests.size > 0 && @total_size > @count.to_i %>
     <div class="col-sm-4">
-      <%= erb :_paging, locals: { url: "#{root_path}unique_digests" } %>
+      <%= erb unique_template(:_paging), locals: { url: "#{root_path}unique_digests" } %>
     </div>
   <% end %>
 </header>

redis/create.lua

@@ -45,10 +45,4 @@ end
 redis.call('GETSET', version_key, api_version)
 redis.call('PERSIST', exists_key)
 
-if expiration then
-  redis.call('EXPIRE', available_key, expiration)
-  redis.call('EXPIRE', exists_key, expiration)
-  redis.call('EXPIRE', version_key, expiration)
-end
-
 return job_id

redis/expire.lua

@@ -0,0 +1,14 @@
+-- redis.replicate_commands();
+
+local exists_key    = KEYS[1]
+local available_key = KEYS[2]
+local version_key   = KEYS[3]
+
+local expiration    = tonumber(ARGV[1])
+
+if expiration then
+  redis.log(redis.LOG_DEBUG, "create.lua - expiring locks because expiration: " .. tostring(expiration))
+  redis.call('EXPIRE', available_key, expiration)
+  redis.call('EXPIRE', exists_key, expiration)
+  redis.call('EXPIRE', version_key, expiration)
+end

redis/signal.lua

@@ -9,7 +9,7 @@ local token      = ARGV[1]
 local expiration = tonumber(ARGV[2])
 
 redis.call('HDEL', grabbed_key, token)
-redis.call('SREM', unique_keys, token)
+redis.call('SREM', unique_keys, unique_digest)
 local available_count = redis.call('LPUSH', available_key, token)
 
 if expiration then
@@ -21,3 +21,4 @@ if expiration then
 end
 
 return available_count
+

spec/integration/sidekiq_unique_jobs/lock/until_expired_spec.rb

@@ -45,6 +45,8 @@
       process_one.lock
       process_one.execute {}
       expect(process_one.locked?).to eq(true)
+      expect(ttl('uniquejobs:da6005926a8457526e998f0033901dfc:EXISTS')).to eq(1)
+      expect(ttl('uniquejobs:da6005926a8457526e998f0033901dfc:VERSION')).to eq(1)
     end
   end
 

spec/integration/sidekiq_unique_jobs/locksmith_spec.rb

@@ -9,7 +9,7 @@
   let(:jid_one)         { 'maaaahjid' }
   let(:jid_two)           { 'jidmayhem' }
   let(:lock_expiration) { nil }
-  let(:unique_digest)   { 'test_mutex_key' }
+  let(:unique_digest)   { 'uniquejobs:randomvalue' }
   let(:item_one) do
     {
       'jid' => jid_one,
@@ -115,6 +115,44 @@
 
     it_behaves_like 'a lock'
 
+    it 'creates the expected keys' do
+      locksmith_one.lock
+      expect(unique_digests).to match_array(['uniquejobs:randomvalue'])
+      expect(unique_keys).to match_array(%w[
+                                           uniquejobs:randomvalue:EXISTS
+                                           uniquejobs:randomvalue:GRABBED
+                                           uniquejobs:randomvalue:VERSION
+                                         ])
+    end
+
+    it 'expires the expected keys' do
+      locksmith_one.lock
+      expect(unique_digests).to match_array(['uniquejobs:randomvalue'])
+      expect(unique_keys).to match_array(%w[
+                                           uniquejobs:randomvalue:EXISTS
+                                           uniquejobs:randomvalue:GRABBED
+                                           uniquejobs:randomvalue:VERSION
+                                         ])
+      locksmith_one.signal
+
+      expect(unique_digests).to match_array([])
+      expect(ttl('uniquejobs:randomvalue:EXISTS')).to eq(1)
+      expect(ttl('uniquejobs:randomvalue:VERSION')).to eq(1)
+    end
+
+    it 'deletes the expected keys' do
+      locksmith_one.lock
+      expect(unique_digests).to match_array(['uniquejobs:randomvalue'])
+      expect(unique_keys).to match_array(%w[
+                                           uniquejobs:randomvalue:EXISTS
+                                           uniquejobs:randomvalue:GRABBED
+                                           uniquejobs:randomvalue:VERSION
+                                         ])
+      locksmith_one.delete!
+      expect(unique_digests).to match_array([])
+      expect(unique_keys).to match_array(%w[])
+    end
+
     it 'expires keys' do
       Sidekiq.redis(&:flushdb)
       locksmith_one.create

spec/integration/sidekiq_unique_jobs/web_spec.rb

@@ -30,6 +30,15 @@ def app
     expect(last_response.body).to match("/unique_digests/#{another_digest}")
   end
 
+  it 'can paginate digests' do
+    110.times do |idx|
+      expect(MyUniqueJob.perform_async(1, idx)).not_to eq(nil)
+    end
+
+    get '/unique_digests'
+    expect(last_response.status).to eq(200)
+  end
+
   it 'can display digest' do
     expect(MyUniqueJob.perform_async(1, 2)).not_to eq(nil)
 

spec/support/sidekiq_helpers.rb

@@ -55,6 +55,14 @@ def ttl(key)
     redis { |conn| conn.ttl(key) }
   end
 
+  def unique_digests
+    smembers('unique:keys')
+  end
+
+  def smembers(key)
+    redis { |conn| conn.smembers(key) }
+  end
+
   def unique_keys
     keys('uniquejobs:*')
   end