diff --git a/test/test_cli_jobs.rb b/test/test_cli_jobs.rb index ea46253..4903743 100644 --- a/test/test_cli_jobs.rb +++ b/test/test_cli_jobs.rb @@ -156,3 +156,93 @@ class CliJobsTest < Minitest::Test queue.enqueue(job_class: 'CliJobsEchoJob', args: ['after-stale-pid']) @cli.jobs_work(['--once', "--pidfile=#{pidfile}"]) + assert_equal ['after-stale-pid'], CliJobsEchoJob.log + refute File.exist?(pidfile) + end + + # Real end-to-end daemon smoke test: runs `bin/prism jobs work` as its own + # OS process (no --once, no --max-cycles), lets it poll and drain a job, + # then sends it SIGINT the same way an operator or init system would and + # confirms it shuts down cleanly rather than needing to be killed. + def test_daemon_process_drains_jobs_and_shuts_down_cleanly_on_sigint + job_file = File.join(@dir, 'daemon_job.rb') + marker_path = File.join(@dir, 'daemon_marker.txt') + File.write(job_file, <<~RUBY) + class CliJobsDaemonJob + def perform(marker_path) + File.write(marker_path, 'ran') + end + end + RUBY + queue.enqueue(job_class: 'CliJobsDaemonJob', args: [marker_path]) + + bin = File.expand_path('../bin/prism', __dir__) + pid = Process.spawn( + bin, 'jobs', 'work', '--interval=0.05', '--require=daemon_job.rb', + chdir: @dir, out: File::NULL, err: File::NULL + ) + + deadline = Time.now + 5 + sleep 0.05 until File.exist?(marker_path) || Time.now > deadline + assert File.exist?(marker_path), 'daemon did not process the job within 5s' + assert_equal 'ran', File.read(marker_path) + + Process.kill('INT', pid) + _, status = Process.wait2(pid) + assert status.success?, "daemon did not exit cleanly after SIGINT (status: #{status})" + end + + # Real horizontal-scaling scenario: two independent `bin/prism jobs work` + # daemons (separate OS processes, no pidfile since neither is meant to be + # "the only one") polling the SAME queue file. Jobs::Queue#claim_next! is + # already proven race-free at the storage layer (test_jobs_queue.rb spawns + # 4 bare claimer processes), but that does not by itself prove the CLI + # daemon loop (poll -> work_off -> sleep -> repeat, plus its own error + # handling/verbose logging) is safe to run in more than one copy - this + # drives the actual `bin/prism jobs work` entrypoint twice and checks both + # that every job runs exactly once AND that work genuinely got split + # between the two processes rather than one daemon draining the whole + # queue before the other gets a chance to poll. + def test_two_real_daemon_processes_share_the_same_queue_without_dropping_or_duplicating_jobs + job_count = 60 + results_dir = File.join(@dir, 'daemon_results') + FileUtils.mkdir_p(results_dir) + + job_file = File.join(@dir, 'multi_daemon_job.rb') + File.write(job_file, <<~RUBY) + class CliJobsMultiDaemonJob + def perform(id, results_dir) + sleep 0.02 + File.write(File.join(results_dir, "job-\#{id}.done"), Process.pid.to_s) + end + end + RUBY + + job_count.times { |i| queue.enqueue(job_class: 'CliJobsMultiDaemonJob', args: [i, results_dir]) } + + bin = File.expand_path('../bin/prism', __dir__) + pids = 2.times.map do + Process.spawn( + bin, 'jobs', 'work', '--interval=0.01', '--require=multi_daemon_job.rb', + chdir: @dir, out: File::NULL, err: File::NULL + ) + end + + deadline = Time.now + 10 + sleep 0.05 while queue.size.positive? && Time.now < deadline + + pids.each { |pid| Process.kill('INT', pid) } + pids.each { |pid| Process.wait2(pid) } + + assert_equal 0, queue.size, 'both daemons together should have drained the whole queue within the deadline' + + result_files = Dir.glob(File.join(results_dir, 'job-*.done')) + assert_equal job_count, result_files.size, 'every job should have run exactly once (no drops, no duplicates)' + expected_names = (0...job_count).map { |i| "job-#{i}.done" }.sort + assert_equal expected_names, result_files.map { |f| File.basename(f) }.sort + + processing_pids = result_files.map { |f| File.read(f).strip }.uniq + assert_equal pids.map(&:to_s).sort, processing_pids.sort, + 'both daemon processes should have actually processed jobs, not just one of them' + end +end diff --git a/test/test_cli_jobs_concurrent_daemons.rb b/test/test_cli_jobs_concurrent_daemons.rb new file mode 100644 index 0000000..80ddfcf --- /dev/null +++ b/test/test_cli_jobs_concurrent_daemons.rb @@ -0,0 +1,74 @@ +require 'minitest/autorun' +require 'prism' +require 'tmpdir' +require 'fileutils' + +# The unit-level race (Queue#claim_next! across raw concurrent processes, +# test_jobs_queue.rb) and the single-daemon pidfile guard (test_cli_jobs.rb) +# were both already covered, but nothing exercised the shape a real +# horizontally-scaled deploy actually takes: several `prism jobs work` +# daemons - each going through the FULL Cli#jobs_work -> Jobs::Worker -> +# job#perform -> mark_completed pipeline, each with its OWN pidfile so the +# single-daemon guard never gets in their way - pointed at the SAME queue +# file at the same time. +class CliJobsConcurrentDaemonsTest < Minitest::Test + DAEMON_COUNT = 5 + JOB_COUNT = 60 + + def setup + @dir = Dir.mktmpdir('prism-cli-jobs-concurrent') + @queue_path = File.join(@dir, 'db', 'jobs.pstore') + @claims_dir = File.join(@dir, 'claims') + @pids_dir = File.join(@dir, 'pids') + FileUtils.mkdir_p(@claims_dir) + FileUtils.mkdir_p(@pids_dir) + end + + def teardown + FileUtils.remove_entry(@dir) + end + + def test_concurrent_daemons_each_process_every_job_exactly_once_with_no_lost_or_duplicated_work + queue = Prism::Jobs::Queue.new(@queue_path) + markers = JOB_COUNT.times.map(&:to_s) + markers.each { |marker| queue.enqueue(job_class: 'CliJobsConcurrentClaimJob', args: [marker]) } + + job_class_file = File.join(@dir, 'claim_job.rb') + File.write(job_class_file, <<~RUBY) + class CliJobsConcurrentClaimJob + def perform(marker) + sleep(rand * 0.01) + File.write(File.join(#{@claims_dir.inspect}, "\#{marker}.claim"), Process.pid.to_s) + end + end + RUBY + + daemon_script = File.join(@dir, 'daemon.rb') + File.write(daemon_script, <<~RUBY) + require 'prism' + cli = Prism::Cli.new(#{@dir.inspect}) + cli.jobs_work( + ['--once', '--file=#{@queue_path}', '--require=#{job_class_file}', "--pidfile=\#{ARGV[0]}"], + out: File.open(File::NULL, 'w') + ) + RUBY + + pids = DAEMON_COUNT.times.map do |i| + pidfile = File.join(@pids_dir, "worker-#{i}.pid") + Process.spawn(RbConfig.ruby, '-I', File.expand_path('../lib', __dir__), daemon_script, pidfile, + out: File::NULL, err: File::NULL) + end + pids.each { |pid| Process.waitpid(pid) } + + claim_files = Dir.glob(File.join(@claims_dir, '*.claim')) + claimed_markers = claim_files.map { |f| File.basename(f, '.claim') } + + assert_equal JOB_COUNT, claim_files.size, 'every enqueued job should have been performed exactly once' + assert_equal markers.sort, claimed_markers.sort, 'the exact set of enqueued jobs should have been claimed, no more, no less' + assert_equal claimed_markers.size, claimed_markers.uniq.size, 'no job should have been performed twice by two different daemons' + + assert_equal 0, queue.size, 'the queue should be fully drained' + assert_equal 0, queue.dead_letter_size, 'no job should have failed' + assert_empty Dir.glob(File.join(@pids_dir, '*.pid')), 'every daemon should have removed its own pidfile on exit' + end +end