Allow defining a job priority [implements #56]

lib/rihanna/job.ex

@@ -67,7 +67,8 @@ defmodule Rihanna.Job do
     :enqueued_at,
     :due_at,
     :failed_at,
-    :fail_reason
+    :fail_reason,
+    :priority
   ]
 
   defstruct @fields
@@ -87,6 +88,14 @@ defmodule Rihanna.Job do
     GenServer.call(Rihanna.JobManager, job)
   end
 
+  @doc """
+  The priority of this job.
+
+  Conforms to the niceness values used in Linux processes — lower values
+  are more important.
+  """
+  def priority, do: 19
+
   @doc false
   def enqueue(term, due_at \\ nil) do
     serialized_term = :erlang.term_to_binary(term)
@@ -96,11 +105,11 @@ defmodule Rihanna.Job do
     result =
       producer_query(
         """
-          INSERT INTO "#{table()}" (term, enqueued_at, due_at)
-          VALUES ($1, $2, $3)
+          INSERT INTO "#{table()}" (term, enqueued_at, due_at, priority)
+          VALUES ($1, $2, $3, $4)
           RETURNING #{@sql_fields}
         """,
-        [serialized_term, now, due_at]
+        [serialized_term, now, due_at, priority()]
       )
 
     case result do
@@ -134,15 +143,17 @@ defmodule Rihanna.Job do
         enqueued_at,
         due_at,
         failed_at,
-        fail_reason
+        fail_reason,
+        priority
       ]) do
     %__MODULE__{
       id: id,
       term: :erlang.binary_to_term(serialized_term),
       enqueued_at: enqueued_at,
       due_at: due_at,
       failed_at: failed_at,
-      fail_reason: fail_reason
+      fail_reason: fail_reason,
+      priority: priority
     }
   end
 
@@ -252,7 +263,7 @@ defmodule Rihanna.Job do
           WHERE NOT (id = ANY($3))
           AND (due_at IS NULL OR due_at <= now())
           AND failed_at IS NULL
-          ORDER BY enqueued_at, j.id
+          ORDER BY priority, enqueued_at, j.id
           FOR UPDATE OF j SKIP LOCKED
           LIMIT 1
         ) AS t1
@@ -266,7 +277,7 @@ defmodule Rihanna.Job do
               AND (due_at IS NULL OR due_at <= now())
               AND failed_at IS NULL
               AND (j.enqueued_at, j.id) > (jobs.enqueued_at, jobs.id)
-              ORDER BY enqueued_at, j.id
+              ORDER BY priority, enqueued_at, j.id
               FOR UPDATE OF j SKIP LOCKED
               LIMIT 1
             ) AS j

lib/rihanna/migration.ex

@@ -104,6 +104,7 @@ defmodule Rihanna.Migration do
       CREATE TABLE #{table_name} (
         id int NOT NULL,
         term bytea NOT NULL,
+        priority integer NOT NULL DEFAULT 19,
         enqueued_at timestamp with time zone NOT NULL,
         due_at timestamp with time zone,
         failed_at timestamp with time zone,
@@ -135,7 +136,7 @@ defmodule Rihanna.Migration do
       ADD CONSTRAINT #{table_name}_pkey PRIMARY KEY (id);
       """,
       """
-      CREATE INDEX #{table_name}_enqueued_at_id ON #{table_name} (enqueued_at ASC, id ASC);
+      CREATE INDEX #{table_name}_enqueued_at_id ON #{table_name} (priority ASC, enqueued_at ASC, id ASC);
       """
     ]
   end

lib/rihanna/migration/upgrade.ex

@@ -72,6 +72,17 @@ defmodule Rihanna.Migration.Upgrade do
       """,
       """
       ALTER TABLE #{table_name} DROP COLUMN rihanna_internal_meta;
+      """,
+      """
+      ALTER TABLE #{table_name} DROP COLUMN priority;
+      """,
+      """
+      DO $$
+          BEGIN
+              DROP INDEX IF EXISTS rihanna_jobs_enqueued_at_id;
+              CREATE INDEX IF NOT EXISTS rihanna_jobs_enqueued_at_id ON rihanna_jobs (enqueued_at ASC, id ASC);
+          END;
+      $$
       """
     ]
   end
@@ -116,6 +127,17 @@ defmodule Rihanna.Migration.Upgrade do
       """,
       """
       CREATE INDEX IF NOT EXISTS rihanna_jobs_enqueued_at_id ON rihanna_jobs (enqueued_at ASC, id ASC);
+      """,
+      """
+      ALTER TABLE #{table_name} ADD COLUMN priority integer NOT NULL DEFAULT 19;
+      """,
+      """
+      DO $$
+          BEGIN
+              DROP INDEX IF EXISTS rihanna_jobs_enqueued_at_id;
+              CREATE INDEX IF NOT EXISTS rihanna_jobs_enqueued_at_id ON rihanna_jobs (priority ASC, enqueued_at ASC, id ASC);
+          END;
+      $$
       """
     ]
   end

test/rihanna/job_test.exs

@@ -89,7 +89,7 @@ defmodule Rihanna.JobTest do
         [job] ++
           [
             insert_job(pg, :ready_to_run),
-            insert_job(pg, :ready_to_run)
+            insert_job(pg, :ready_to_run_highest_priority)
           ]
 
       {:ok, %{jobs: jobs}}
@@ -102,6 +102,12 @@ defmodule Rihanna.JobTest do
       assert length(locked) == 3
     end
 
+    test "locks all available jobs, ordered with the highest priority first", %{pg: pg} do
+      [ first_job | _rest ] = lock(pg, 4)
+
+      assert %Rihanna.Job{priority: -19} = first_job
+    end
+
     test "locks all available jobs if equal to N", %{pg: pg, jobs: jobs} do
       locked = lock(pg, 3)
 

test/rihanna_test.exs

@@ -20,6 +20,7 @@ defmodule RihannaTest do
       assert job.due_at |> is_nil
       assert job.fail_reason |> is_nil
       assert job.failed_at |> is_nil
+      assert job.priority == 19
       assert job.term == @term
     end
 
@@ -33,6 +34,7 @@ defmodule RihannaTest do
       assert job.due_at |> is_nil
       assert job.fail_reason |> is_nil
       assert job.failed_at |> is_nil
+      assert job.priority == 19
       assert job.term == @term
     end
 
@@ -67,6 +69,7 @@ defmodule RihannaTest do
       assert job.due_at |> is_nil
       assert job.fail_reason |> is_nil
       assert job.failed_at |> is_nil
+      assert job.priority == 19
       assert job.term == {Rihanna.Mocks.MockJob, :arg}
     end
 
@@ -80,6 +83,7 @@ defmodule RihannaTest do
       assert job.due_at |> is_nil
       assert job.fail_reason |> is_nil
       assert job.failed_at |> is_nil
+      assert job.priority == 19
       assert job.term == {Rihanna.Mocks.MockJob, :arg}
     end
 
@@ -122,6 +126,7 @@ defmodule RihannaTest do
           assert job.due_at |> is_nil
           assert job.fail_reason |> is_nil
           assert job.failed_at |> is_nil
+          assert job.priority == 19
           assert job.term == {Rihanna.Mocks.MockJob, :arg}
 
           assert {:ok, _} = Rihanna.delete(job.id)
@@ -144,6 +149,7 @@ defmodule RihannaTest do
         assert job.due_at |> is_nil
         assert job.fail_reason |> is_nil
         assert job.failed_at |> is_nil
+        assert job.priority == 19
         assert job.term == {Rihanna.Mocks.MockJob, :arg}
 
         assert {:ok, _} = Rihanna.delete(job.id)
@@ -166,6 +172,7 @@ defmodule RihannaTest do
                 fail_reason: nil,
                 failed_at: nil,
                 id: 1,
+                priority: 19,
                 term: {Rihanna.Mocks.MockJob, :arg}
               }} = Rihanna.delete(job.id)
 

test/support/test_helper.ex

@@ -61,7 +61,7 @@ defmodule TestHelper do
       exec.(
         conn,
         """
-        SELECT id, term, enqueued_at, due_at, failed_at, fail_reason FROM "rihanna_jobs" WHERE id = $1
+        SELECT id, term, enqueued_at, due_at, failed_at, fail_reason, priority FROM "rihanna_jobs" WHERE id = $1
         """,
         [id]
       )
@@ -81,7 +81,24 @@ defmodule TestHelper do
         """
           INSERT INTO "rihanna_jobs" (term, enqueued_at)
           VALUES ($1, '2018-01-01')
-          RETURNING id, term, enqueued_at, due_at, failed_at, fail_reason
+          RETURNING id, term, enqueued_at, due_at, failed_at, fail_reason, priority
+        """,
+        [:erlang.term_to_binary(@test_term)]
+      )
+
+    [job] = Rihanna.Job.from_sql(result.rows)
+
+    job
+  end
+
+  def insert_job(pg, :ready_to_run_highest_priority) do
+    result =
+      Postgrex.query!(
+        pg,
+        """
+          INSERT INTO "rihanna_jobs" (term, enqueued_at, priority)
+          VALUES ($1, '2018-01-01', -19)
+          RETURNING id, term, enqueued_at, due_at, failed_at, fail_reason, priority
         """,
         [:erlang.term_to_binary(@test_term)]
       )
@@ -99,7 +116,7 @@ defmodule TestHelper do
         """
           INSERT INTO "rihanna_jobs" (term, enqueued_at, due_at)
           VALUES ($1, '2018-01-01', now() + interval '1 minute')
-          RETURNING id, term, enqueued_at, due_at, failed_at, fail_reason
+          RETURNING id, term, enqueued_at, due_at, failed_at, fail_reason, priority
         """,
         [:erlang.term_to_binary(@test_term)]
       )
@@ -117,7 +134,7 @@ defmodule TestHelper do
         """
           INSERT INTO "rihanna_jobs" (term, enqueued_at, due_at)
           VALUES ($1, '2018-01-01', now())
-          RETURNING id, term, enqueued_at, due_at, failed_at, fail_reason
+          RETURNING id, term, enqueued_at, due_at, failed_at, fail_reason, priority
         """,
         [:erlang.term_to_binary(@test_term)]
       )
@@ -139,7 +156,7 @@ defmodule TestHelper do
           fail_reason
         )
         VALUES ($1, '2018-01-01', '2018-01-02', 'Kaboom!')
-        RETURNING id, term, enqueued_at, due_at, failed_at, fail_reason
+        RETURNING id, term, enqueued_at, due_at, failed_at, fail_reason, priority
         """,
         [:erlang.term_to_binary(@test_term)]
       )