Fix unique-job lock leak in inline testing mode

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

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

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