Make snooze and unique middleware play nice with each other

Unique middleware unlocks the keys even if the job is snoozed and not
completed. Make it aware of the snooze middleware.

Author: ananthakumaran

README.md

@@ -558,7 +558,9 @@ Exq comes with an `enqueue_all` method which guarantees atomicity.
 
 A Job can be snoozed by returning `{:snooze, time_to_sleep_in_seconds}`
 from perform method. By default this feature is not enabled. Add
-`Exq.Middleware.Snooze` to the middleware list to enable this feature.
+`Exq.Middleware.Snooze` to the middleware list to enable this
+feature. If you use `Exq.Middleware.Unique`, it must be placed
+before it.
 
 ```elixir
 defmodule MyWorker do

config/test.exs

@@ -26,8 +26,8 @@ config :exq,
     Exq.Middleware.Stats,
     Exq.Middleware.Job,
     Exq.Middleware.Manager,
+    Exq.Middleware.Snooze,
     Exq.Middleware.Unique,
-    Exq.Middleware.Telemetry,
-    Exq.Middleware.Snooze
+    Exq.Middleware.Telemetry
   ],
   queue_adapter: Exq.Adapters.Queue.Mock

lib/exq/middleware/snooze.ex

@@ -2,6 +2,7 @@ defmodule Exq.Middleware.Snooze do
   @behaviour Exq.Middleware.Behaviour
   alias Exq.Redis.JobQueue
   alias Exq.Middleware.Pipeline
+  import Pipeline
 
   def before_work(pipeline) do
     pipeline
@@ -13,15 +14,19 @@ defmodule Exq.Middleware.Snooze do
       )
       when is_number(seconds) do
     if assigns.job do
-      JobQueue.snooze_job(
-        assigns.redis,
-        assigns.namespace,
-        assigns.job,
-        seconds
-      )
-    end
+      {:ok, _jid} =
+        JobQueue.snooze_job(
+          assigns.redis,
+          assigns.namespace,
+          assigns.job,
+          seconds
+        )
 
-    pipeline
+      pipeline
+      |> assign(:job_snoozed, true)
+    else
+      pipeline
+    end
   end
 
   def after_processed_work(pipeline) do

lib/exq/middleware/unique.ex

@@ -21,6 +21,10 @@ defmodule Exq.Middleware.Unique do
     pipeline
   end
 
+  def after_processed_work(%Pipeline{assigns: %{job_snoozed: true}} = pipeline) do
+    pipeline
+  end
+
   def after_processed_work(
         %Pipeline{assigns: %{job_serialized: job_serialized, redis: redis, namespace: namespace}} =
           pipeline

test/config_test.exs

@@ -314,9 +314,9 @@ defmodule Exq.ConfigTest do
              Exq.Middleware.Stats,
              Exq.Middleware.Job,
              Exq.Middleware.Manager,
+             Exq.Middleware.Snooze,
              Exq.Middleware.Unique,
-             Exq.Middleware.Telemetry,
-             Exq.Middleware.Snooze
+             Exq.Middleware.Telemetry
            ]
 
     assert mode == :default

test/exq_test.exs

@@ -814,6 +814,39 @@ defmodule ExqTest do
     stop_process(sup)
   end
 
+  test "snooze + unique job" do
+    Process.register(self(), :exqtest)
+    {:ok, sup} = Exq.start_link(concurrency: 1, queues: ["q1"], scheduler_enable: true)
+
+    {:ok, j1} =
+      Exq.enqueue(Exq, "q1", ExqTest.SnoozeWorker, [0.050, :snoozed],
+        max_retries: 0,
+        unique_for: 500,
+        unique_token: "t1"
+      )
+
+    {:conflict, ^j1} =
+      Exq.enqueue(Exq, "q1", ExqTest.SnoozeWorker, [0.050, :snoozed],
+        max_retries: 0,
+        unique_for: 500,
+        unique_token: "t1"
+      )
+
+    :timer.sleep(100)
+    assert_received {"snoozed"}
+
+    {:conflict, ^j1} =
+      Exq.enqueue(Exq, "q1", ExqTest.SnoozeWorker, [0.050, :snoozed],
+        max_retries: 0,
+        unique_for: 500,
+        unique_token: "t1"
+      )
+
+    :timer.sleep(100)
+    assert_received {"snoozed"}
+    stop_process(sup)
+  end
+
   test "cancel job" do
     {:ok, sup} = Exq.start_link()
     {:ok, _} = Exq.enqueue(Exq, "default", ExqTest.SleepWorker, [60 * 1000, :worked])

test/middleware_test.exs

@@ -239,9 +239,9 @@ defmodule MiddlewareTest do
       Exq.Middleware.Stats,
       Exq.Middleware.Job,
       Exq.Middleware.Manager,
+      Exq.Middleware.Snooze,
       Exq.Middleware.Unique,
-      Exq.Middleware.Telemetry,
-      Exq.Middleware.Snooze
+      Exq.Middleware.Telemetry
     ]
 
     assert Middleware.all(Middleware) == chain