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