Add a descriptive name field to the job (#15)

* Add a descriptive name field to the job

This name does not have to be unique, and allows easier lookup of the
job or a specific set of jobs.

* changelog

* update name in recurring job data

* update changelog

* cleanup

effectum/CHANGELOG.md

@@ -1,3 +1,7 @@
+# 0.6.0
+
+- Jobs can now have a name, which can be used as an argument to `get_jobs_by_name`. The name does not have to be unique.
+
 # 0.5.1
 
 - Allow configuring whether to use the `Display` or `Debug` formatter to format failure information.

effectum/examples/recurring.rs

@@ -3,18 +3,13 @@
 
 use std::{
     path::Path,
-    sync::{
-        atomic::{AtomicI64, Ordering},
-        Arc, Mutex,
-    },
+    sync::{Arc, Mutex},
     time::Duration,
 };
 
 use ahash::HashMap;
 use effectum::{Job, JobRunner, Queue, RecurringJobSchedule, RunningJob};
-use serde::{Deserialize, Serialize};
 use time::OffsetDateTime;
-use tracing::{event, Level};
 use tracing_subscriber::{layer::SubscriberExt, EnvFilter};
 
 #[derive(Debug)]

effectum/migrations/00003-job-name-column.sql

@@ -0,0 +1,6 @@
+ALTER TABLE jobs
+  ADD COLUMN name text;
+
+CREATE INDEX jobs_name_added_at ON jobs (name, added_at DESC)
+WHERE
+  name IS NOT NULL;

effectum/src/add_job.rs

@@ -28,6 +28,9 @@ pub struct Job {
     pub id: Uuid,
     /// The name of the job, which matches the name used in the [JobRunner](crate::JobRunner) for the job.
     pub job_type: Cow<'static, str>,
+    /// A description for this job which can be passed to [Queue::get_jobs_by_name]. This value does not
+    /// have to be unique among all jobs.
+    pub name: Option<String>,
     /// Jobs with higher `priority` will be executed first.
     pub priority: i32,
     /// Jobs that are expected to take more processing resources can be given a higher weight
@@ -98,6 +101,7 @@ impl Default for Job {
         Self {
             id: Uuid::now_v7(),
             job_type: Default::default(),
+            name: None,
             priority: 0,
             weight: 1,
             run_at: Default::default(),
@@ -126,6 +130,20 @@ impl JobBuilder {
         }
     }
 
+    /// Set the name of this job. This name is purely informational, and does not have to be unique. Jobs can be fetched by
+    /// their name using [Queue::get_jobs_by_name].
+    pub fn name(mut self, name: impl ToString) -> Self {
+        self.job.name = Some(name.to_string());
+        self
+    }
+
+    /// Set the name of this job. This name is purely informational, and does not have to be unique. Jobs can be fetched by
+    /// their name using [Queue::get_jobs_by_name].
+    pub fn name_opt(mut self, name: Option<String>) -> Self {
+        self.job.name = name.map(|n| n.to_string());
+        self
+    }
+
     /// Set the priority of the job.
     pub fn priority(mut self, priority: i32) -> Self {
         self.job.priority = priority;

effectum/src/db_writer/add_job.rs

@@ -24,11 +24,11 @@ pub(crate) struct AddMultipleJobsArgs {
 
 pub(super) const INSERT_JOBS_QUERY: &str = r##"
     INSERT INTO jobs
-    (external_id, job_type, status, priority, weight, from_base_job, orig_run_at, payload,
+    (external_id, job_type, name, status, priority, weight, from_base_job, orig_run_at, payload,
         max_retries, backoff_multiplier, backoff_randomization, backoff_initial_interval,
         added_at, default_timeout, heartbeat_increment, run_info)
     VALUES
-    ($external_id, $job_type, $status, $priority, $weight, $from_base_job, $run_at, $payload,
+    ($external_id, $job_type, $name, $status, $priority, $weight, $from_base_job, $run_at, $payload,
         $max_retries, $backoff_multiplier, $backoff_randomization, $backoff_initial_interval,
         $added_at, $default_timeout, $heartbeat_increment, '[]')
 "##;
@@ -52,6 +52,7 @@ pub(super) fn execute_add_job_stmt(
     jobs_stmt.execute(named_params! {
         "$external_id": &job_config.id,
         "$job_type": job_config.job_type,
+        "$name": job_config.name,
         "$priority": job_config.priority,
         "$weight": job_config.weight,
         "$from_base_job": job_config.from_recurring,

effectum/src/db_writer/ready_jobs.rs

@@ -51,7 +51,8 @@ fn do_get_ready_jobs(
                 backoff_randomization,
                 backoff_initial_interval,
                 max_retries,
-                orig_run_at
+                orig_run_at,
+                jobs.name
             FROM active_jobs
             JOIN jobs USING(job_id)
             WHERE active_worker_id IS NULL
@@ -66,6 +67,7 @@ fn do_get_ready_jobs(
     struct JobResult {
         job_id: i64,
         external_id: Uuid,
+        name: Option<String>,
         priority: i32,
         weight: u16,
         job_type: String,
@@ -103,6 +105,7 @@ fn do_get_ready_jobs(
             let backoff_initial_interval: i32 = row.get(11)?;
             let max_retries: i32 = row.get(12)?;
             let orig_run_at: i64 = row.get(13)?;
+            let name: Option<String> = row.get(14)?;
 
             Ok(JobResult {
                 job_id,
@@ -119,6 +122,7 @@ fn do_get_ready_jobs(
                 backoff_initial_interval,
                 max_retries,
                 orig_run_at,
+                name,
             })
         },
     )?;
@@ -160,6 +164,7 @@ fn do_get_ready_jobs(
         let job = RunningJob(Arc::new(RunningJobData {
             id: job.external_id,
             job_id: job.job_id,
+            name: job.name,
             worker_id,
             heartbeat_increment: job.heartbeat_increment,
             job_type: job.job_type,

effectum/src/db_writer/recurring.rs

@@ -219,7 +219,8 @@ fn update_existing_recurring_job(
             backoff_randomization = ?8,
             backoff_initial_interval = ?9,
             default_timeout = ?10,
-            heartbeat_increment = ?11
+            heartbeat_increment = ?11,
+            name = ?12
         WHERE job_id=?1"##,
     )?;
     base_update_stmt.execute(params![
@@ -234,6 +235,7 @@ fn update_existing_recurring_job(
         job.retries.backoff_initial_interval.as_secs(),
         job.timeout.as_secs(),
         job.heartbeat_increment.as_secs(),
+        job.name,
     ])?;
 
     // Update any pending jobs
@@ -250,7 +252,8 @@ fn update_existing_recurring_job(
             backoff_randomization = ?,
             backoff_initial_interval = ?,
             default_timeout = ?,
-            heartbeat_increment = ?
+            heartbeat_increment = ?,
+            name = ?
         WHERE from_base_job = ? AND status = 'pending'
         RETURNING job_id"##,
     )?;
@@ -269,6 +272,7 @@ fn update_existing_recurring_job(
                 job.retries.backoff_initial_interval.as_secs(),
                 job.timeout.as_secs(),
                 job.heartbeat_increment.as_secs(),
+                job.name,
                 base_job_id,
             ],
             |row| row.get::<_, rusqlite::types::Value>(0),

effectum/src/job.rs

@@ -47,6 +47,8 @@ pub struct RunningJobData {
     /// The id of this job.
     pub id: Uuid,
     pub(crate) job_id: i64,
+    /// The name given to this job
+    pub name: Option<String>,
     /// The ID of the [Worker](crate::worker::Worker) that is running this job.
     pub worker_id: WorkerId,
     /// How many seconds a heartbeat can extend the expiration time.
@@ -83,6 +85,7 @@ impl Debug for RunningJobData {
         f.debug_struct("Job")
             .field("id", &self.id)
             .field("job_id", &self.job_id)
+            .field("name", &self.name)
             .field("worker_id", &self.worker_id)
             .field("heartbeat_increment", &self.heartbeat_increment)
             .field("job_type", &self.job_type)
@@ -111,8 +114,14 @@ impl Display for RunningJobData {
 
         write!(
             f,
-            "Job {{ id: {}, job_type: {}, priority: {}, start_time: {}, expires: {}, try: {} }}",
-            self.id, self.job_type, self.priority, self.start_time, expires, self.current_try
+            "Job {{ id: {}, name: {}, job_type: {}, priority: {}, start_time: {}, expires: {}, try: {} }}",
+            self.id,
+            self.name.as_deref().unwrap_or("_"),
+            self.job_type,
+            self.priority,
+            self.start_time,
+            expires,
+            self.current_try
         )
     }
 }