Set dedupe lock timeout on CacheDeploySpecJob

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

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