Fixed so that backing off after failures is respected by other workers

lib/delayed/job.rb

@@ -165,7 +165,7 @@ def lock_exclusively!(max_run_time, worker = worker_name)
       now = self.class.db_time_now
       affected_rows = if locked_by != worker
         # We don't own this job so we will update the locked_by name and the locked_at
-        self.class.update_all(["locked_at = ?, locked_by = ?", now, worker], ["id = ? and (locked_at is null or locked_at < ?)", id, (now - max_run_time.to_i)])
+        self.class.update_all(["locked_at = ?, locked_by = ?", now, worker], ["id = ? and (locked_at is null or locked_at < ?) and (run_at <= ?)", id, (now - max_run_time.to_i), now])
       else
         # We already own this job, this may happen if the job queue crashes.
         # Simply resume and update the locked_at

spec/job_spec.rb

@@ -218,8 +218,27 @@ def perform; @@runs += 1; end
       @job.lock_exclusively! 5.minutes, 'worker1'
       @job.lock_exclusively! 5.minutes, 'worker1'
     end                                        
-  end            
+  end
 
+  context "when another worker has worked on a task since the job was found to be available, it" do
+
+    before :each do
+      Delayed::Job.worker_name = 'worker1'
+      @job = Delayed::Job.create :payload_object => SimpleJob.new
+      @job_copy_for_worker_2 = Delayed::Job.find(@job.id)
+    end
+
+    it "should not allow a second worker to get exclusive access if already successfully processed by worker1" do
+      @job.delete
+      @job_copy_for_worker_2.lock_exclusively!(4.hours, 'worker2').should == false
+    end
+
+    it "should not allow a second worker to get exclusive access if failed to be processed by worker1 and run_at time is now in future (due to backing off behaviour)" do
+      @job.update_attributes(:attempts => 1, :run_at => Time.now + 1.day)
+      @job_copy_for_worker_2.lock_exclusively!(4.hours, 'worker2').should == false
+    end
+  end
+
   context "#name" do
     it "should be the class name of the job that was enqueued" do
       Delayed::Job.create(:payload_object => ErrorJob.new ).name.should == 'ErrorJob'