diff --git a/lib/sidekiq_unique_jobs/lua/shared/_delete_from_sorted_set.lua b/lib/sidekiq_unique_jobs/lua/shared/_delete_from_sorted_set.lua index 78f53cab4..b035ca525 100644 --- a/lib/sidekiq_unique_jobs/lua/shared/_delete_from_sorted_set.lua +++ b/lib/sidekiq_unique_jobs/lua/shared/_delete_from_sorted_set.lua @@ -1,19 +1,22 @@ local function delete_from_sorted_set(name, digest) - local per = 50 - local total = redis.call("zcard", name) - local index = 0 - local result + local cursor = "0" + local result = nil + local match_pattern = "*" .. digest .. "*" - while (index < total) do - local items = redis.call("ZRANGE", name, index, index + per -1) - for _, item in pairs(items) do + repeat + local scan_result = redis.call("ZSCAN", name, cursor, "MATCH", match_pattern, "COUNT", 1) + cursor = scan_result[1] + local items = scan_result[2] + + for i = 1, #items, 2 do + local item = items[i] if string.find(item, digest) then redis.call("ZREM", name, item) result = item break end end - index = index + per - end + until cursor == "0" or result ~= nil + return result end diff --git a/spec/performance/unique_job_on_conflict_replace_spec.rb b/spec/performance/unique_job_on_conflict_replace_spec.rb new file mode 100644 index 000000000..b643e22a3 --- /dev/null +++ b/spec/performance/unique_job_on_conflict_replace_spec.rb @@ -0,0 +1,57 @@ +# frozen_string_literal: true + +RSpec.describe UniqueJobOnConflictReplace, :perf do + let(:lock_prefix) { SidekiqUniqueJobs.config.lock_prefix } + let(:lock_timeout) { SidekiqUniqueJobs.config.lock_timeout } + let(:lock_ttl) { SidekiqUniqueJobs.config.lock_ttl } + let(:queue) { described_class.sidekiq_options["queue"] } + let(:on_conflict) { described_class.sidekiq_options["on_conflict"] } + let(:lock) { described_class.sidekiq_options["lock"] } + + before do + digests.delete_by_pattern("*") + end + + context "when schedule queue is large" do + it "locks and replaces quickly" do + (0..100_000).each_slice(1_000) do |nums| + redis do |conn| + conn.pipelined do |pipeline| + nums.each do |num| + created_at = Time.now.to_f + scheduled_at = created_at + rand(3_600..2_592_000) + + payload = { + "retry" => true, + "queue" => queue, + "lock" => lock, + "on_conflict" => on_conflict, + "class" => described_class.name, + "args" => [num, { "type" => "extremely unique" }], + "jid" => SecureRandom.hex(12), + "created_at" => created_at, + "lock_timeout" => lock_timeout, + "lock_ttl" => lock_ttl, + "lock_prefix" => lock_prefix, + "lock_args" => [num, { "type" => "extremely unique" }], + "lock_digest" => "#{lock_prefix}:#{SecureRandom.hex}" + } + + pipeline.zadd("schedule", scheduled_at, payload.to_json) + end + end + end + end + + # queueing it once at the end of the queue should succeed + expect(described_class.perform_in(2_592_000, 100_000, { "type" => "extremely unique" })).not_to be_nil + + # queueing it again should be performant + expect do + Timeout.timeout(0.1) do + described_class.perform_in(2_592_000, 100_000, { "type" => "extremely unique" }) + end + end.not_to raise_error + end + end +end diff --git a/spec/support/workers/unique_job_on_conflict_replace.rb b/spec/support/workers/unique_job_on_conflict_replace.rb new file mode 100644 index 000000000..64c6a5631 --- /dev/null +++ b/spec/support/workers/unique_job_on_conflict_replace.rb @@ -0,0 +1,14 @@ +# frozen_string_literal: true + +# :nocov: + +class UniqueJobOnConflictReplace + include Sidekiq::Worker + sidekiq_options lock: :until_executing, + queue: :customqueue, + on_conflict: :replace + + def perform(one, two) + [one, two] + end +end diff --git a/spec/workers/unique_job_on_conflict_replace_spec.rb b/spec/workers/unique_job_on_conflict_replace_spec.rb new file mode 100644 index 000000000..6571ef440 --- /dev/null +++ b/spec/workers/unique_job_on_conflict_replace_spec.rb @@ -0,0 +1,18 @@ +# frozen_string_literal: true + +RSpec.describe UniqueJobOnConflictReplace do + it_behaves_like "sidekiq with options" do + let(:options) do + { + "lock" => :until_executing, + "on_conflict" => :replace, + "queue" => :customqueue, + "retry" => true, + } + end + end + + it_behaves_like "a performing worker" do + let(:args) { ["hundred", { "type" => "extremely unique", "id" => 44 }] } + end +end