feat: return EnqueueOutcome instead of error for duplicate job keys

When queuing a job with a duplicate id, as you might do if you
are queuing jobs with deterministic ids, subsequent attempts will
fail, and add_job() will respond with a row not found error,
because it fails to find any newly-added job.

We now catch this error and handle it gracefully. The EnqueueOutcome
enum lets us distinguish between the case of this job being added
successfully, and it not being added because it was a duplicate.

pub enum EnqueueOutcome {
    // success gives you the job
    Enqueued(Box<Job>),
    // duplicate gives you the key back
    AlreadyInProgress { job_key: String },
}

This is arguably a bug in graphile_worker: it should use
fetch_optional(), not fetch_one(). OTOH, maybe it's intended.

This is a breaking change. Instead of `Result<Job, Error>`,
`enqueue()` now returns `Result<EnqueueOutcome, Error>`.
The upside is you can do idempotent job enqueueing patterns
where duplicate submissions are normal and not errors. This
happens to be the use case I wrote this library for, so yay.

Author: ceejbot

Cargo.lock

@@ -97,11 +97,14 @@ async fn send_notification(
     };
 
     match state.backfill.enqueue("send_email", &email_req, spec).await {
-        Ok(job) => Ok(Json(serde_json::json!({
-            "status": "enqueued",
-            "job_id": job.id(),
-            "message": format!("Email queued for {}", email_req.to)
-        }))),
+        Ok(outcome) => {
+            let job = outcome.unwrap();
+            Ok(Json(serde_json::json!({
+                "status": "enqueued",
+                "job_id": job.id(),
+                "message": format!("Email queued for {}", email_req.to)
+            })))
+        }
         Err(e) => {
             error!("Failed to enqueue email: {}", e);
             Err(StatusCode::INTERNAL_SERVER_ERROR)

examples/admin_server.rs

@@ -77,7 +77,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         should_fail: Some(false),
     };
 
-    let job = client
+    let outcome = client
         .enqueue(
             "example_job",
             &example_job,
@@ -92,7 +92,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     println!(
         "✉️  Enqueued ExampleJob: {} (job_id: {})",
         example_job.message,
-        job.id()
+        outcome.unwrap().id()
     );
 
     // Enqueue a high-priority email job using the convenience function
@@ -103,14 +103,18 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         template: Some("welcome".to_string()),
     };
 
-    let job = enqueue_fast(
+    let outcome = enqueue_fast(
         &client,
         "send_email",
         &email_job,
         Some("welcome-email-user123".to_string()),
     )
     .await?;
-    println!("📧 Enqueued SendEmailJob to: {} (job_id: {})", email_job.to, job.id());
+    println!(
+        "📧 Enqueued SendEmailJob to: {} (job_id: {})",
+        email_job.to,
+        outcome.unwrap().id()
+    );
 
     // Enqueue a bulk processing job
     let process_job = ProcessUserDataJob {
@@ -119,7 +123,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         batch_size: Some(100),
     };
 
-    let job = enqueue_bulk(
+    let outcome = enqueue_bulk(
         &client,
         "process_user_data",
         &process_job,
@@ -129,7 +133,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     println!(
         "📊 Enqueued ProcessUserDataJob for user: {} (job_id: {})",
         process_job.user_id,
-        job.id()
+        outcome.unwrap().id()
     );
 
     // Enqueue a scheduled report generation job (delayed by 30 seconds)
@@ -140,7 +144,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         recipients: vec!["admin@example.com".to_string(), "manager@example.com".to_string()],
     };
 
-    let job = client
+    let outcome = client
         .enqueue(
             "generate_report",
             &report_job,
@@ -156,7 +160,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     println!(
         "📈 Enqueued GenerateReportJob (scheduled for 30s): {} report (job_id: {})",
         report_job.report_type,
-        job.id()
+        outcome.unwrap().id()
     );
 
     // Enqueue a job that will fail (for testing error handling)
@@ -166,7 +170,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         should_fail: Some(true),
     };
 
-    let job = client
+    let outcome = client
         .enqueue(
             "example_job",
             &failing_job,
@@ -180,7 +184,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         .await?;
     println!(
         "💥 Enqueued failing ExampleJob for error testing (job_id: {})",
-        job.id()
+        outcome.unwrap().id()
     );
 
     println!("\n🔄 Demonstrating exponential backoff retry policies...");
@@ -194,7 +198,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         template: Some("alert".to_string()),
     };
 
-    let job = enqueue_critical(
+    let outcome = enqueue_critical(
         &client,
         "send_email",
         &critical_job,
@@ -203,7 +207,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     .await?;
     println!(
         "🚨 Enqueued critical alert with aggressive retries (job_id: {})",
-        job.id()
+        outcome.unwrap().id()
     );
 
     // Enqueue a job with fast retries for quick turnaround
@@ -213,7 +217,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         should_fail: Some(false),
     };
 
-    let job = enqueue_fast_with_retries(
+    let outcome = enqueue_fast_with_retries(
         &client,
         "example_job",
         &notification_job,
@@ -222,7 +226,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     .await?;
     println!(
         "⚡ Enqueued fast notification with quick retries (job_id: {})",
-        job.id()
+        outcome.unwrap().id()
     );
 
     // Enqueue a bulk job with conservative retries
@@ -232,14 +236,17 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         batch_size: Some(1000),
     };
 
-    let job = enqueue_bulk_with_retries(
+    let outcome = enqueue_bulk_with_retries(
         &client,
         "process_user_data",
         &bulk_job,
         Some("bulk-export-789".to_string()),
     )
     .await?;
-    println!("📦 Enqueued bulk job with conservative retries (job_id: {})", job.id());
+    println!(
+        "📦 Enqueued bulk job with conservative retries (job_id: {})",
+        outcome.unwrap().id()
+    );
 
     // Enqueue a job with custom retry policy
     let custom_retry_policy = RetryPolicy::new(
@@ -257,7 +264,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         recipients: vec!["data-team@example.com".to_string()],
     };
 
-    let job = client
+    let outcome = client
         .enqueue(
             "generate_report",
             &custom_job,
@@ -272,7 +279,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         .await?;
     println!(
         "📊 Enqueued analytics job with custom retry policy (job_id: {})",
-        job.id()
+        outcome.unwrap().id()
     );
 
     println!("\n🎯 All jobs enqueued successfully!");

examples/enqueue_jobs.rs

@@ -257,7 +257,7 @@ where
     }
 
     match client.enqueue(&req.task_identifier, &req.payload, spec).await {
-        Ok(job) => {
+        Ok(crate::EnqueueOutcome::Enqueued(job)) => {
             let response = EnqueueJobResponse {
                 job_id: job.id().to_string(),
                 status: "enqueued".to_string(),
@@ -267,6 +267,17 @@ where
             info!("Successfully enqueued job: id={}", job.id());
             Ok(Json(response))
         }
+        Ok(crate::EnqueueOutcome::AlreadyInProgress { job_key }) => {
+            // Return 409 Conflict for already in progress
+            warn!("Job already in progress: job_key={}", job_key);
+            Err((
+                StatusCode::CONFLICT,
+                Json(ErrorResponse::new(
+                    format!("Job with key '{}' is already in progress", job_key),
+                    "JOB_ALREADY_IN_PROGRESS",
+                )),
+            ))
+        }
         Err(e) => {
             error!("Failed to enqueue job: {}", e);
             Err((

src/admin.rs

@@ -329,10 +329,20 @@ impl BackfillClient {
         };
 
         // Enqueue the job
-        let job = self
+        let outcome = self
             .enqueue(&dlq_job.task_identifier, &dlq_job.payload, spec.clone())
             .await?;
 
+        let job = match outcome {
+            crate::EnqueueOutcome::Enqueued(job) => job,
+            crate::EnqueueOutcome::AlreadyInProgress { job_key } => {
+                return Err(BackfillError::RuntimeError(format!(
+                    "Cannot requeue DLQ job {}: a job with key '{}' is already in progress",
+                    dlq_id, job_key
+                )));
+            }
+        };
+
         // Record metrics
         crate::metrics::record_dlq_job_requeued(&dlq_job.task_identifier, spec.queue.as_str());
 
@@ -361,7 +371,7 @@ impl BackfillClient {
             .execute(&self.pool)
             .await?;
 
-        Ok(job)
+        Ok(*job)
     }
 
     /// Delete a job from the DLQ permanently.

src/client/dlq.rs

@@ -1,12 +1,20 @@
 //! The queue functions implementations for backfill client.
 
-use graphile_worker::{Job, TaskHandler, WorkerUtils};
+use graphile_worker::{TaskHandler, WorkerUtils};
 use serde::Serialize;
 use sqlx::PgPool;
 use sqlx::postgres::PgPoolOptions;
 
 use super::BackfillClient;
-use crate::{BackfillError, JobSpec};
+use crate::{BackfillError, EnqueueOutcome, JobSpec};
+
+/// Check if a GraphileWorkerError is a RowNotFound error.
+/// This happens when add_job returns NULL because a job with the same
+/// job_key is currently locked by a worker.
+fn is_row_not_found(e: &graphile_worker::errors::GraphileWorkerError) -> bool {
+    use graphile_worker::errors::GraphileWorkerError;
+    matches!(e, GraphileWorkerError::SqlError(sqlx::Error::RowNotFound))
+}
 
 impl BackfillClient {
     /// Create a new BackfillClient with the given database URL.
@@ -67,13 +75,21 @@ impl BackfillClient {
     /// * `spec` - Job specification including priority, scheduling, etc.
     ///
     /// # Returns
-    /// The Job struct containing the job ID and metadata.
-    pub async fn enqueue<T>(&self, task_identifier: &str, payload: &T, spec: JobSpec) -> Result<Job, BackfillError>
+    /// `EnqueueOutcome::Enqueued(Job)` if the job was created or updated,
+    /// `EnqueueOutcome::AlreadyInProgress { job_key }` if a job with the same
+    /// key is currently being processed.
+    pub async fn enqueue<T>(
+        &self,
+        task_identifier: &str,
+        payload: &T,
+        spec: JobSpec,
+    ) -> Result<EnqueueOutcome, BackfillError>
     where
         T: Serialize,
     {
         let start = std::time::Instant::now();
         let utils = self.utils();
+        let job_key = spec.job_key.clone();
 
         let result = utils
             .add_raw_job(task_identifier, serde_json::to_value(payload)?, spec.clone().into())
@@ -96,12 +112,35 @@ impl BackfillClient {
                     spec.priority.0
                 );
 
-                Ok(job)
+                Ok(EnqueueOutcome::Enqueued(Box::new(job)))
             }
             Err(e) => {
-                crate::metrics::record_db_operation("enqueue", "error");
-                log::error!("Failed to enqueue job (task: {}, error: {})", task_identifier, e);
-                Err(e.into())
+                // Check if this is a RowNotFound error with a job_key
+                // This indicates the job_key is already locked by a worker
+                if is_row_not_found(&e) && job_key.is_some() {
+                    // Job with this key is already locked/in progress
+                    crate::metrics::record_db_operation("enqueue", "already_in_progress");
+                    crate::metrics::record_db_operation_duration("enqueue", start.elapsed().as_secs_f64());
+
+                    // Record the already_in_progress metric
+                    crate::metrics::record_job_already_in_progress(spec.queue.as_str(), task_identifier);
+
+                    let key = job_key.clone().unwrap_or_else(|| "<unknown>".to_string());
+                    log::debug!(
+                        "Job already in progress (job_key: {}, task: {}, queue: {})",
+                        key,
+                        task_identifier,
+                        spec.queue.as_str()
+                    );
+
+                    Ok(EnqueueOutcome::AlreadyInProgress {
+                        job_key: job_key.unwrap_or_default(),
+                    })
+                } else {
+                    crate::metrics::record_db_operation("enqueue", "error");
+                    log::error!("Failed to enqueue job (task: {}, error: {})", task_identifier, e);
+                    Err(e.into())
+                }
             }
         }
     }
@@ -110,13 +149,20 @@ impl BackfillClient {
     ///
     /// This method uses the task's IDENTIFIER constant and ensures the payload
     /// type matches the expected task type.
-    pub async fn enqueue_task<T>(&self, task: T, spec: JobSpec) -> Result<Job, BackfillError>
+    pub async fn enqueue_task<T>(&self, task: T, spec: JobSpec) -> Result<EnqueueOutcome, BackfillError>
     where
         T: TaskHandler + Serialize,
     {
         let utils = self.utils();
-        let job = utils.add_job(task, spec.into()).await?;
-        Ok(job)
+        let job_key = spec.job_key.clone();
+
+        match utils.add_job(task, spec.into()).await {
+            Ok(job) => Ok(EnqueueOutcome::Enqueued(Box::new(job))),
+            Err(e) if is_row_not_found(&e) && job_key.is_some() => Ok(EnqueueOutcome::AlreadyInProgress {
+                job_key: job_key.unwrap_or_default(),
+            }),
+            Err(e) => Err(e.into()),
+        }
     }
 
     /// Remove a job by its unique key.