diff --git a/app/jobs/shipit/cache_deploy_spec_job.rb b/app/jobs/shipit/cache_deploy_spec_job.rb index 6ae032397..b9e5aad91 100644 --- a/app/jobs/shipit/cache_deploy_spec_job.rb +++ b/app/jobs/shipit/cache_deploy_spec_job.rb @@ -7,13 +7,25 @@ class CacheDeploySpecJob < BackgroundJob queue_as :deploys + # Caps job execution AND sets the dedupe lock expiration to match. + # Without it the lock falls back to Unique::DEFAULT_TIMEOUT (10s), which is + # far shorter than the job's runtime, letting duplicate jobs for the same + # stack run concurrently once the lock expires mid-run. + self.timeout = 15.minutes.to_i + def perform(stack) return if stack.inaccessible? + commit = stack.commits.reachable.last commands = Commands.for(stack) - commands.with_temporary_working_directory(commit: stack.commits.reachable.last, recursive: false) do |path| + commands.with_temporary_working_directory(commit:, recursive: false) do |path| stack.update!(cached_deploy_spec: DeploySpec::FileSystem.new(path, stack)) end + + # A duplicate enqueued while this job held the dedupe lock was dropped; + # if the head moved under us, that dropped job's work is still + # outstanding, so hand it off rather than leaving the spec stale. + CacheDeploySpecJob.perform_later(stack) if stack.commits.reachable.last&.id != commit&.id end end end diff --git a/test/jobs/cache_deploy_spec_job_test.rb b/test/jobs/cache_deploy_spec_job_test.rb index 95d696bce..5e71ef05c 100644 --- a/test/jobs/cache_deploy_spec_job_test.rb +++ b/test/jobs/cache_deploy_spec_job_test.rb @@ -21,5 +21,60 @@ class CacheDeploySpecJobTest < ActiveSupport::TestCase @job.perform(@stack) assert_equal [], @stack.reload.checklist end + + test "the dedupe lock expiration covers the job runtime" do + assert_operator CacheDeploySpecJob.timeout, :>, BackgroundJob::Unique::DEFAULT_TIMEOUT + assert_equal 15.minutes.to_i, CacheDeploySpecJob.timeout + end + + test "the redis lock is created with the job timeout as its expiration" do + mutex = mock + mutex.expects(:lock).yields + Redis::Lock.expects(:new) + .with(anything, anything, expiration: 15.minutes.to_i, timeout: 0) + .returns(mutex) + + executed = false + CacheDeploySpecJob.new(@stack).acquire_lock { executed = true } + assert executed + end + + test "#perform re-enqueues itself when the head moves during the run" do + moved_head = @stack.commits.reachable.first + reachable = mock + reachable.stubs(:last).returns(@last_commit, moved_head) + @stack.stubs(:commits).returns(stub(reachable:)) + @stack.stubs(:update!) # side-effect callbacks are irrelevant to this test + + StackCommands.any_instance.expects(:with_temporary_working_directory) + .with(commit: @last_commit, recursive: false).yields(Pathname(Dir.tmpdir)) + + assert_enqueued_with(job: CacheDeploySpecJob, args: [@stack]) do + @job.perform(@stack) + end + end + + test "#perform does not re-enqueue itself when the head is unchanged" do + StackCommands.any_instance.expects(:with_temporary_working_directory) + .with(commit: @last_commit, recursive: false).yields(Pathname(Dir.tmpdir)) + + assert_no_enqueued_jobs(only: CacheDeploySpecJob) do + @job.perform(@stack) + end + end + + test "a duplicate job for the same stack is dropped while the lock is held" do + job = CacheDeploySpecJob.new(@stack) + duplicate = CacheDeploySpecJob.new(@stack) + duplicate_ran = false + + job.acquire_lock do + duplicate.acquire_lock do + duplicate_ran = true + end + end + + refute duplicate_ran, "duplicate should have been dropped, not executed" + end end end