Add ability to snooze job

Implementation is same as how retry is implemented, just that
retry_count is not incremented, so a job can be snoozed unlimited times

Author: ananthakumaran

README.md

@@ -554,6 +554,20 @@ Exq comes with an `enqueue_all` method which guarantees atomicity.
 ])
 ```
 
+## Snooze
+
+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.
+
+```elixir
+defmodule MyWorker do
+  def perform do
+    {:snooze, 10}
+  end
+end
+```
+
 ## Web UI
 
 Exq has a separate repo, exq_ui which provides with a Web UI to monitor your workers:

config/test.exs

@@ -27,6 +27,7 @@ config :exq,
     Exq.Middleware.Job,
     Exq.Middleware.Manager,
     Exq.Middleware.Unique,
-    Exq.Middleware.Telemetry
+    Exq.Middleware.Telemetry,
+    Exq.Middleware.Snooze
   ],
   queue_adapter: Exq.Adapters.Queue.Mock

lib/exq/middleware/pipeline.ex

@@ -47,7 +47,7 @@ defmodule Exq.Middleware.Pipeline do
   end
 
   @doc """
-  Sets `terminated` to true
+  Sets `terminated` to true
   """
   def terminate(%Pipeline{} = pipeline) do
     %{pipeline | terminated: true}

lib/exq/middleware/snooze.ex

@@ -0,0 +1,34 @@
+defmodule Exq.Middleware.Snooze do
+  @behaviour Exq.Middleware.Behaviour
+  alias Exq.Redis.JobQueue
+  alias Exq.Middleware.Pipeline
+
+  def before_work(pipeline) do
+    pipeline
+  end
+
+  def after_processed_work(
+        %Pipeline{assigns: %{result: {:snooze, seconds}} = assigns} =
+          pipeline
+      )
+      when is_number(seconds) do
+    if assigns.job do
+      JobQueue.snooze_job(
+        assigns.redis,
+        assigns.namespace,
+        assigns.job,
+        seconds
+      )
+    end
+
+    pipeline
+  end
+
+  def after_processed_work(pipeline) do
+    pipeline
+  end
+
+  def after_failed_work(pipeline) do
+    pipeline
+  end
+end

lib/exq/redis/job_queue.ex

@@ -317,6 +317,26 @@ defmodule Exq.Redis.JobQueue do
     end
   end
 
+  def snooze_job(redis, namespace, job, offset) do
+    job =
+      %{job | error_message: "Snoozed for #{offset} seconds"}
+      |> add_failure_timestamp()
+
+    time = Time.offset_from_now(offset)
+    Logger.info("Queueing job #{job.jid} to retry in #{offset} seconds")
+
+    {:ok, _jid} =
+      do_enqueue_job_at(
+        redis,
+        namespace,
+        job,
+        Job.encode(job),
+        job.jid,
+        time,
+        retry_queue_key(namespace)
+      )
+  end
+
   def retry_or_fail_job(redis, namespace, job, error) do
     if dead?(job) do
       Logger.info("Max retries on job #{job.jid} exceeded")

test/config_test.exs

@@ -315,7 +315,8 @@ defmodule Exq.ConfigTest do
              Exq.Middleware.Job,
              Exq.Middleware.Manager,
              Exq.Middleware.Unique,
-             Exq.Middleware.Telemetry
+             Exq.Middleware.Telemetry,
+             Exq.Middleware.Snooze
            ]
 
     assert mode == :default

test/exq_test.exs

@@ -47,6 +47,13 @@ defmodule ExqTest do
     end
   end
 
+  defmodule SnoozeWorker do
+    def perform(time, message) do
+      send(:exqtest, {message})
+      {:snooze, time}
+    end
+  end
+
   setup do
     TestRedis.setup()
 
@@ -766,6 +773,17 @@ defmodule ExqTest do
     stop_process(sup)
   end
 
+  test "snooze job" do
+    Process.register(self(), :exqtest)
+    {:ok, sup} = Exq.start_link(concurrency: 1, queues: ["q1"], scheduler_enable: true)
+    {:ok, _} = Exq.enqueue(Exq, "q1", ExqTest.SnoozeWorker, [0.050, :snoozed], max_retries: 0)
+    :timer.sleep(100)
+    assert_received {"snoozed"}
+    :timer.sleep(100)
+    assert_received {"snoozed"}
+    stop_process(sup)
+  end
+
   defp enqueue_fail_job(count) do
     for _ <- 0..(count - 1) do
       {:ok, _} =

test/job_queue_test.exs

@@ -171,6 +171,37 @@ defmodule JobQueueTest do
     end)
   end
 
+  test "snooze job" do
+    with_application_env(:exq, :max_retries, 1, fn ->
+      JobQueue.snooze_job(
+        :testredis,
+        "test",
+        %{
+          retry_count: 0,
+          retry: true,
+          queue: "default",
+          class: "MyWorker",
+          jid: UUID.uuid4(),
+          error_class: nil,
+          error_message: "failed",
+          retried_at: Time.unix_seconds(),
+          failed_at: Time.unix_seconds(),
+          enqueued_at: Time.unix_seconds(),
+          finished_at: nil,
+          processor: nil,
+          args: [],
+          unique_for: nil,
+          unique_until: nil,
+          unique_token: nil,
+          unlocks_at: nil
+        },
+        10
+      )
+
+      assert JobQueue.queue_size(:testredis, "test", :retry) == 1
+    end)
+  end
+
   test "scheduler_dequeue max_score" do
     add_usecs = fn time, offset ->
       base = time |> DateTime.to_unix(:microsecond)

test/middleware_test.exs

@@ -240,7 +240,8 @@ defmodule MiddlewareTest do
       Exq.Middleware.Job,
       Exq.Middleware.Manager,
       Exq.Middleware.Unique,
-      Exq.Middleware.Telemetry
+      Exq.Middleware.Telemetry,
+      Exq.Middleware.Snooze
     ]
 
     assert Middleware.all(Middleware) == chain