diff --git a/test/test_jobs_queue.rb b/test/test_jobs_queue.rb new file mode 100644 index 0000000..136b574 --- /dev/null +++ b/test/test_jobs_queue.rb @@ -0,0 +1,187 @@ +require 'minitest/autorun' +require 'prism' +require 'tmpdir' +require 'fileutils' + +class JobsQueueTest < Minitest::Test + def setup + @dir = Dir.mktmpdir('prism-jobs-queue') + @queue = Prism::Jobs::Queue.new(File.join(@dir, 'jobs.pstore')) + end + + def teardown + FileUtils.remove_entry(@dir) + end + + def test_enqueue_returns_an_incrementing_id + first = @queue.enqueue(job_class: 'X', args: []) + second = @queue.enqueue(job_class: 'X', args: []) + assert_equal first + 1, second + end + + def test_next_ready_returns_the_earliest_due_pending_job + later = @queue.enqueue(job_class: 'X', args: [1], run_at: Time.now - 5) + sooner = @queue.enqueue(job_class: 'X', args: [2], run_at: Time.now - 50) + job = @queue.next_ready + assert_equal sooner, job[:id] + refute_equal later, job[:id] + end + + def test_jobs_scheduled_in_the_future_are_not_ready + @queue.enqueue(job_class: 'X', args: [], run_at: Time.now + 3600) + assert_nil @queue.next_ready + end + + def test_next_ready_respects_queue_name_filter + @queue.enqueue(job_class: 'X', args: [], queue_name: 'mailers') + critical_id = @queue.enqueue(job_class: 'X', args: [], queue_name: 'critical') + job = @queue.next_ready(queue_name: 'critical') + assert_equal critical_id, job[:id] + end + + def test_claim_next_returns_the_earliest_due_pending_job_and_marks_it_processing + later = @queue.enqueue(job_class: 'X', args: [1], run_at: Time.now - 5) + sooner = @queue.enqueue(job_class: 'X', args: [2], run_at: Time.now - 50) + + job = @queue.claim_next! + assert_equal sooner, job[:id] + refute_equal later, job[:id] + assert_equal 'processing', job[:status] + assert_equal 'processing', @queue.pending_jobs.find { |j| j[:id] == sooner }[:status] + end + + def test_claim_next_never_returns_the_same_job_twice + id = @queue.enqueue(job_class: 'X', args: []) + first = @queue.claim_next! + second = @queue.claim_next! + assert_equal id, first[:id] + assert_nil second + end + + def test_claim_next_respects_queue_name_filter + @queue.enqueue(job_class: 'X', args: [], queue_name: 'mailers') + critical_id = @queue.enqueue(job_class: 'X', args: [], queue_name: 'critical') + job = @queue.claim_next!(queue_name: 'critical') + assert_equal critical_id, job[:id] + end + + def test_claim_next_is_nil_when_nothing_is_ready + @queue.enqueue(job_class: 'X', args: [], run_at: Time.now + 3600) + assert_nil @queue.claim_next! + end + + # Real cross-process race: two OS processes sharing the same PStore file + # both try to grab the same batch of jobs at once. #claim_next! selects + # and flips status to 'processing' inside one flock'd write transaction, + # so the file itself is the arbiter of who wins each job - no job should + # ever be claimed by more than one process. + def test_claim_next_is_safe_across_real_concurrent_processes + queue_path = File.join(@dir, 'jobs.pstore') + job_count = 40 + job_count.times { @queue.enqueue(job_class: 'X', args: []) } + + claims_dir = File.join(@dir, 'claims') + FileUtils.mkdir_p(claims_dir) + claimer_script = File.join(@dir, 'claimer.rb') + File.write(claimer_script, <<~RUBY) + require 'prism' + queue = Prism::Jobs::Queue.new(#{queue_path.inspect}) + claims_dir = #{claims_dir.inspect} + loop do + job = queue.claim_next! + break unless job + File.write(File.join(claims_dir, "\#{job[:id]}.claim"), Process.pid.to_s) + end + RUBY + + pids = 4.times.map do + Process.spawn(RbConfig.ruby, '-I', File.expand_path('../lib', __dir__), claimer_script, + out: File::NULL, err: File::NULL) + end + pids.each { |pid| Process.waitpid(pid) } + + claim_files = Dir.glob(File.join(claims_dir, '*.claim')) + assert_equal job_count, claim_files.size, 'every enqueued job should have been claimed exactly once' + assert_equal claim_files.size, claim_files.map { |f| File.basename(f) }.uniq.size, + 'no job id should have produced more than one claim file' + end + + def test_mark_completed_removes_the_job + id = @queue.enqueue(job_class: 'X', args: []) + @queue.mark_completed(id) + assert_nil @queue.next_ready + assert_equal 0, @queue.size + end + + def test_mark_failed_reschedules_with_exponential_backoff_until_max_attempts + id = @queue.enqueue(job_class: 'X', args: [], max_attempts: 3, backoff: { strategy: :exponential, base_delay: 2 }) + before = Time.now + @queue.mark_failed(id, 'boom') + job = @queue.pending_jobs.first + assert_equal 1, job[:attempts] + assert_equal 'pending', job[:status] + assert_equal 'boom', job[:last_error] + assert_in_delta (before + 2).to_f, job[:run_at].to_f, 1.0 + + @queue.mark_failed(id, 'boom again') + job = @queue.pending_jobs.first + assert_equal 2, job[:attempts] + assert_in_delta (before + 4).to_f, job[:run_at].to_f, 1.0 + end + + def test_mark_failed_dead_letters_after_exhausting_max_attempts + id = @queue.enqueue(job_class: 'X', args: [], max_attempts: 2) + @queue.mark_failed(id, 'first failure') + @queue.mark_failed(id, 'second failure') + + assert_equal 0, @queue.size + assert_equal 1, @queue.dead_letter_size + dead = @queue.dead_jobs.first + assert_equal 'dead', dead[:status] + assert_equal 'second failure', dead[:last_error] + end + + def test_retry_dead_moves_a_dead_job_back_to_pending + id = @queue.enqueue(job_class: 'X', args: [], max_attempts: 1) + @queue.mark_failed(id, 'fatal') + assert_equal 1, @queue.dead_letter_size + + @queue.retry_dead(id) + assert_equal 0, @queue.dead_letter_size + job = @queue.next_ready + assert_equal id, job[:id] + assert_equal 0, job[:attempts] + assert_nil job[:last_error] + end + + def test_linear_backoff_grows_by_a_constant_step + id = @queue.enqueue(job_class: 'X', args: [], max_attempts: 5, backoff: { strategy: :linear, base_delay: 3 }) + before = Time.now + @queue.mark_failed(id, 'e1') + job = @queue.pending_jobs.first + assert_in_delta (before + 3).to_f, job[:run_at].to_f, 1.0 + + @queue.mark_failed(id, 'e2') + job = @queue.pending_jobs.first + assert_in_delta (before + 6).to_f, job[:run_at].to_f, 1.0 + end + + def test_state_survives_reopening_the_store + id = @queue.enqueue(job_class: 'PersistedJob', args: [42]) + path = File.join(@dir, 'jobs.pstore') + reopened = Prism::Jobs::Queue.new(path) + job = reopened.next_ready + assert_equal id, job[:id] + assert_equal [42], job[:args] + end + + def test_clear_empties_both_tables + @queue.enqueue(job_class: 'X', args: []) + dead_id = @queue.enqueue(job_class: 'X', args: [], max_attempts: 1) + @queue.mark_failed(dead_id, 'boom') + + @queue.clear! + assert_equal 0, @queue.size + assert_equal 0, @queue.dead_letter_size + end +end