Additional thread-safety spec

lib/sidekiq/debouncer/enq.rb

@@ -5,7 +5,7 @@ module Debouncer
     class Enq < ::Sidekiq::Scheduled::Enq
       extend LuaCommands
 
-      SET = 'debouncer'
+      SET = "debouncer"
       LUA_ZPOPBYSCORE_WITHSCORE = File.read(File.expand_path("../lua/zpopbyscore_withscore.lua", __FILE__))
       LUA_ZPOPBYSCORE_MULTI = File.read(File.expand_path("../lua/zpopbyscore_multi.lua", __FILE__))
 
@@ -17,9 +17,11 @@ def initialize(config)
           super()
           @client = Sidekiq::Client
           @redis = Sidekiq.method(:redis)
+          @logger = Sidekiq.logger
         else
           super(config)
           @redis = config.method(:redis)
+          @logger = config.logger
         end
       end
 
@@ -34,7 +36,7 @@ def enqueue_jobs
 
             @client.push({"args" => final_args, "class" => klass, "debounce_key" => job})
 
-            logger.debug { "enqueued #{SET}: #{job}" }
+            @logger.debug { "enqueued #{SET}: #{job}" }
           end
         end
       end

spec/sidekiq/debouncer/middleware/client_spec.rb

@@ -203,8 +203,6 @@
 
         expect(queue.size).to eq(0)
       end
-
-      # TODO: add a spec to check if enquing new job while pulling is going to work correctly
     end
 
     context "sidekiq testing fake mode" do

spec/sidekiq/debouncer/middleware/server_spec.rb

@@ -33,4 +33,27 @@
       processor.process_one
     end
   end
+
+  context "multiprocess safety" do
+    it "is safe" do
+      Parallel.each((1..100).to_a, in_processes: 10) do |i|
+        TestWorker.perform_async("A", i)
+      end
+
+      expect(schedule_set.size).to eq(1)
+
+      set_item = schedule_set.first
+      expect(set_item.value).to eq("debounce/TestWorker/A")
+
+      expect(puller.instance_variable_get(:@enq)).to receive(:zpopbyscore_withscore).twice.and_wrap_original do |original_method, *args, **kwargs|
+        original_method.call(*args, **kwargs).tap { TestWorker.perform_async("A", 101) }
+      end
+
+      Timecop.freeze(time_start + 10 * 60)
+      puller.enqueue
+
+      expect(queue.size).to eq(1)
+      expect(queue.first.args.map { _1[1] }).to match_array((1..100).to_a)
+    end
+  end
 end

spec/support/context.rb

@@ -30,5 +30,6 @@
       connection.call("FLUSHDB")
       connection.call("SCRIPT", "FLUSH")
     end
+    sidekiq_config[:error_handlers] << ->(ex, _) { raise ex }
   end
 end