From 62213f87d410bd875d9f48a407a5d1d0f691da52 Mon Sep 17 00:00:00 2001
From: Michael Lovitt <lovitt@castolon.co>
Date: Fri, 26 Jun 2026 09:39:14 -0500
Subject: [PATCH] Fix unique-job lock leak under inline execution mode

The provisional/final lock split added in 0.15.0 assumes the block
passed to `lock_for_scheduling!` only enqueues the job. Under inline
testing mode (`Cloudtasker::Testing.inline!`) the block also executes
the job synchronously, which releases the job's lock via the execute
middleware. Step 3 then re-acquires a "final" lock that nothing will
ever release, so a subsequent enqueue of the same unique job is
wrongly rejected for the full lock TTL.

Skip the final-lock step under inline execution: the job has already
run and released its lock by the time scheduling returns. The async
(production) path is unchanged.

Adds an integration spec (two successive inline enqueues of the same
unique job both run) and a unit spec (the lock is not promoted to the
final TTL in inline mode).
---
 lib/cloudtasker/unique_job/job.rb       | 44 ++++++++++++++++++++-----
 spec/cloudtasker/unique_job/job_spec.rb | 14 ++++++++
 spec/integration/unique_job_spec.rb     | 16 +++++++++
 3 files changed, 65 insertions(+), 9 deletions(-)

diff --git a/lib/cloudtasker/unique_job/job.rb b/lib/cloudtasker/unique_job/job.rb
index d72ca85..3db6f89 100644
--- a/lib/cloudtasker/unique_job/job.rb
+++ b/lib/cloudtasker/unique_job/job.rb
@@ -230,7 +230,35 @@ def lock_for_scheduling!
         # Step 2: Yield to perform scheduling operation
         result = yield
 
-        # Step 3: Set final lock
+        # Step 3: Promote the provisional lock to the final lock
+        #
+        # Skipped under inline execution (testing): in that mode the job runs
+        # synchronously within the yield above and releases its own lock via the
+        # execute middleware. Re-acquiring the lock here would resurrect a lock
+        # that nothing will ever release, wrongly rejecting subsequent enqueues
+        # of the same unique job.
+        set_final_lock! unless inline_mode?
+
+        # Return the result of the block
+        result
+      end
+
+      #
+      # Delete the job lock.
+      #
+      def unlock!
+        locked_id = redis.get(unique_gid)
+        redis.del(unique_gid) if locked_id == id
+      end
+
+      private
+
+      #
+      # Promote the provisional lock to the full lock TTL after the job has been
+      # enqueued. The job is already enqueued at this point, so a failure to
+      # acquire is logged rather than raised.
+      #
+      def set_final_lock!
         # Check if the lock is still held by this job
         acquired = redis.get(unique_gid) == id
 
@@ -240,19 +268,17 @@ def lock_for_scheduling!
         final_lock_acquired = redis.set(unique_gid, id, nx: !acquired, ex: lock_ttl)
 
         # Log a warning if final lock could not be acquired
-        # The job has already been enqueued at this point, so raising an error is useless
         worker.logger.warn(LOCK_FINALIZATION_WARNING) unless final_lock_acquired
-
-        # Return the result of the block
-        result
       end
 
       #
-      # Delete the job lock.
+      # Return true in inline testing mode, where jobs run synchronously at
+      # enqueue time. Mirrors `Backend::MemoryTask.inline_mode?`.
       #
-      def unlock!
-        locked_id = redis.get(unique_gid)
-        redis.del(unique_gid) if locked_id == id
+      # @return [Boolean] True if inline mode enabled.
+      #
+      def inline_mode?
+        defined?(Cloudtasker::Testing) && Cloudtasker::Testing.inline?
       end
     end
   end
diff --git a/spec/cloudtasker/unique_job/job_spec.rb b/spec/cloudtasker/unique_job/job_spec.rb
index 4d34d07..968a8b4 100644
--- a/spec/cloudtasker/unique_job/job_spec.rb
+++ b/spec/cloudtasker/unique_job/job_spec.rb
@@ -378,6 +378,20 @@
         expect(job.redis.ttl(job.unique_gid)).to be_within(2).of(job.lock_provisional_ttl)
       end
     end
+
+    context 'with inline execution mode' do
+      # In inline mode the job runs synchronously within the yield and releases
+      # its own lock, so the final lock must not be re-acquired (nothing would
+      # release it). The TTL therefore stays at the short provisional value.
+      it 'does not promote the lock to the final TTL' do
+        Cloudtasker::Testing.inline! do
+          job.lock_for_scheduling! { block_executed << true }
+        end
+
+        expect(block_executed).to eq([true])
+        expect(job.redis.ttl(job.unique_gid)).to be_within(2).of(job.lock_provisional_ttl)
+      end
+    end
   end
 
   describe '#unlock!' do
diff --git a/spec/integration/unique_job_spec.rb b/spec/integration/unique_job_spec.rb
index 9f3c417..0657d29 100644
--- a/spec/integration/unique_job_spec.rb
+++ b/spec/integration/unique_job_spec.rb
@@ -65,6 +65,22 @@
     end
   end
 
+  describe 'successive inline enqueuing+run of standalone jobs with same args' do
+    before do
+      Cloudtasker::Testing.inline! do
+        TestUniqueJobWorker.perform_async(1, 2)
+        TestUniqueJobWorker.perform_async(1, 2)
+      end
+    end
+
+    # In inline mode the job runs synchronously within the enqueue call and
+    # releases its lock. A second enqueue of the same unique job must therefore
+    # run again rather than being rejected by a lock left behind by the first.
+    it 'processes both jobs' do
+      expect(TestUniqueJobWorker.past_job_args).to eq([[1, 2], [1, 2]])
+    end
+  end
+
   describe 'concurrent enqueuing of a standalone job and batch sub-job (lock_per_batch: true)' do
     before do
       orig_options = TestUniqueJobWorker.cloudtasker_options_hash