Only lock unique lock if it's acquired by the given job

Currently any job can unlock unique token, even if it is set by
different job.

1) j1 sets lock (expires)
2) j2 sets lock
3) j1 unlocks the lock set by j2

This creates more problem, as more jobs can be enqueued with same
unique token. Be conservative and only unlock if the lock is acquired
by the same jid. Incase of expired lock, unlock will become a no-op
instead of unlock lock set by another jid.

Author: ananthakumaran

lib/exq/middleware/unique.ex

@@ -10,9 +10,9 @@ defmodule Exq.Middleware.Unique do
     job = Exq.Support.Job.decode(job_serialized)
 
     case job do
-      %{unique_until: "start", unique_token: unique_token, retry_count: retry_count}
+      %{unique_until: "start", unique_token: unique_token, retry_count: retry_count, jid: jid}
       when retry_count in [0, nil] ->
-        {:ok, _} = JobQueue.unlock(redis, namespace, unique_token)
+        {:ok, _} = JobQueue.unlock(redis, namespace, unique_token, jid)
 
       _ ->
         :ok
@@ -28,8 +28,8 @@ defmodule Exq.Middleware.Unique do
     job = Exq.Support.Job.decode(job_serialized)
 
     case job do
-      %{unique_until: "success", unique_token: unique_token} ->
-        {:ok, _} = JobQueue.unlock(redis, namespace, unique_token)
+      %{unique_until: "success", unique_token: unique_token, jid: jid} ->
+        {:ok, _} = JobQueue.unlock(redis, namespace, unique_token, jid)
 
       _ ->
         :ok
@@ -45,9 +45,9 @@ defmodule Exq.Middleware.Unique do
     job = Exq.Support.Job.decode(job_serialized)
 
     case job do
-      %{unique_until: "success", unique_token: unique_token} ->
+      %{unique_until: "success", unique_token: unique_token, jid: jid} ->
         if JobQueue.dead?(job) do
-          {:ok, _} = JobQueue.unlock(redis, namespace, unique_token)
+          {:ok, _} = JobQueue.unlock(redis, namespace, unique_token, jid)
         end
 
       _ ->

lib/exq/redis/job_queue.ex

@@ -58,14 +58,25 @@ defmodule Exq.Redis.JobQueue do
   end
 
   def unlock_jobs(redis, namespace, raw_jobs) do
-    for job <- raw_jobs,
-        unique_token = Job.decode(job).unique_token do
-      unlock(redis, namespace, unique_token)
+    for raw_job <- raw_jobs,
+        job = Job.decode(raw_job),
+        unique_token = job.unique_token do
+      unlock(redis, namespace, unique_token, job.jid)
     end
   end
 
-  def unlock(redis, namespace, unique_token) do
-    Connection.del!(redis, unique_key(namespace, unique_token), retry_on_connection_error: 3)
+  def unlock(redis, namespace, unique_token, jid) do
+    Exq.Support.Redis.with_retry_on_connection_error(
+      fn ->
+        Script.eval!(
+          redis,
+          :compare_and_delete,
+          [unique_key(namespace, unique_token)],
+          [jid]
+        )
+      end,
+      3
+    )
   end
 
   def enqueue_in(redis, namespace, queue, offset, worker, args, options)

lib/exq/redis/script.ex

@@ -12,6 +12,17 @@ defmodule Exq.Redis.Script do
   end
 
   @scripts %{
+    compare_and_delete:
+      Prepare.script("""
+      local key = KEYS[1]
+      local expected_value = ARGV[1]
+      local current_value = redis.call("get", key)
+      if current_value == expected_value then
+        return redis.call("del", key)
+      else
+        return 0
+      end
+      """),
     enqueue:
       Prepare.script("""
       local queues_key, job_queue_key, unique_key = KEYS[1], KEYS[2], KEYS[3]

test/exq_test.exs

@@ -599,6 +599,36 @@ defmodule ExqTest do
     stop_process(sup)
   end
 
+  test "handle lock expiry gracefully" do
+    Process.register(self(), :exqtest)
+    {:ok, sup} = Exq.start_link(concurrency: 1, queues: ["q1"])
+
+    {:ok, j1} =
+      Exq.enqueue(Exq, "q1", ExqTest.SleepWorker, [3000, :worked1],
+        unique_for: 1,
+        unique_token: "t1"
+      )
+
+    :timer.sleep(2000)
+
+    {:ok, j2} =
+      Exq.enqueue(Exq, "q1", ExqTest.SleepWorker, [3000, :worked],
+        unique_for: 60,
+        unique_token: "t1"
+      )
+
+    :timer.sleep(1500)
+    assert_received {"worked1"}
+
+    {:conflict, ^j2} =
+      Exq.enqueue(Exq, "q1", ExqTest.SleepWorker, [3000, :worked],
+        unique_for: 60,
+        unique_token: "t1"
+      )
+
+    stop_process(sup)
+  end
+
   defmodule ConstantBackoff do
     @behaviour Exq.Backoff.Behaviour