Check timeout on acquire

* Avoid to touch created_at because it is misused and maybe used in
  different meanings in the future. From v0.8.48, PQ can use timeout
  to find living tasks.
* Check timeout on heartbeat. v0.9's AcquiredTask#timeout saves the last
  heartbeated time. PQ can check whether the heartbeating task is still
  acquired by itself or not.

Author: nurse

lib/perfectqueue/backend/rdb.rb

@@ -6,6 +6,7 @@ module PerfectQueue::Backend
   class RDBBackend
     MAX_RETRY = ::PerfectQueue::Backend::RDBCompatBackend::MAX_RETRY
     DELETE_OFFSET = ::PerfectQueue::Backend::RDBCompatBackend::DELETE_OFFSET
+    EVENT_HORIZON = ::PerfectQueue::Backend::RDBCompatBackend::EVENT_HORIZON
     class Token < Struct.new(:key)
     end
 
@@ -48,7 +49,7 @@ def submit(id, data, time=Process.clock_gettime(Process::CLOCK_REALTIME, :second
 
     def cancel(id, delete_timeout=3600, now=Process.clock_gettime(Process::CLOCK_REALTIME, :second))
       connect {
-        n = @db["UPDATE `#{@table}` SET timeout=?, created_at=NULL, resource=NULL WHERE id=? AND created_at IS NOT NULL;", now+delete_timeout-DELETE_OFFSET, id].update
+        n = @db["UPDATE `#{@table}` SET timeout=?, created_at=NULL, resource=NULL WHERE id=? AND #{EVENT_HORIZON} < timeout;", now+delete_timeout-DELETE_OFFSET, id].update
         return n > 0
       }
     end

lib/perfectqueue/backend/rdb_compat.rb

@@ -296,7 +296,7 @@ def finish(task_token, retention_time, options)
         key = task_token.key
 
         connect {
-          n = @db["UPDATE `#{@table}` SET timeout=?, created_at=NULL, resource=NULL WHERE id=? AND created_at IS NOT NULL", delete_timeout, key].update
+          n = @db["UPDATE `#{@table}` SET timeout=?, created_at=NULL, resource=NULL WHERE id=? AND #{EVENT_HORIZON} < timeout", delete_timeout, key].update
           if n <= 0
             raise IdempotentAlreadyFinishedError, "task key=#{key} does not exist or already finished."
           end
@@ -317,8 +317,14 @@ def heartbeat(task_token, alive_time, options)
           sql << ", data=?"
           params << compress_data(data.to_json, options[:compression])
         end
-        sql << " WHERE id=? AND created_at IS NOT NULL"
-        params << key
+        if last_timeout = options[:last_timeout]
+          sql << " WHERE id=? AND timeout=?"
+          params << key
+          params << last_timeout
+        else
+          sql << " WHERE id=? AND #{EVENT_HORIZON} < timeout"
+          params << key
+        end
 
         connect {
           n = @db[*params].update
@@ -327,7 +333,9 @@ def heartbeat(task_token, alive_time, options)
             if row == nil
               raise PreemptedError, "task key=#{key} does not exist or preempted."
             elsif row[:created_at] == nil
-              raise PreemptedError, "task key=#{key} preempted."
+              raise PreemptedError, "task key=#{key} is finished or canceled"
+            elsif options[:last_timeout] && row[:timeout] != options[:last_timeout]
+              raise PreemptedError, "task key=#{key} is preempted by another worker."
             else # row[:timeout] == next_timeout
               # ok
             end