fix: preserve job payloads when moving failed jobs to DLQ

Author: ceejbot

docs/implementation/incomplete-features.md

@@ -147,29 +147,7 @@ let queue_name = "default";
 - Query GraphileWorker schema to get queue info
 - Check if newer graphile_worker versions expose this field
 
-### 9. DLQ payload limitation
-**Location:** `src/client/dlq.rs:542-543`
-
-**Current State:** Uses empty JSON object `{}` as placeholder for payload when moving failed jobs to DLQ
-
-**Issue:** GraphileWorker's jobs view doesn't include the payload field
-
-**Code:**
-```rust
-// Payload is not available in the jobs view, use empty object as placeholder
-let payload = serde_json::json!({});
-```
-
-**Affects:** `process_failed_jobs()` method - jobs moved to DLQ will have empty payloads
-
-**Impact:** Cannot inspect actual job payload from DLQ, only metadata
-
-**Possible Solutions:**
-- Query the underlying jobs table directly (not the view)
-- Store payload in a separate field during initial job processing
-- Accept this limitation and document it clearly
-
-### 10. DLQ pagination optimization
+### 9. DLQ pagination optimization
 **Location:** `src/client/dlq.rs:238`
 
 **Current State:** Uses separate COUNT query for pagination, could be more efficient
@@ -253,8 +231,8 @@ pub(crate) struct JobHandlerConfig {
 - All feature-gated, so not blocking main library use
 
 **Low Priority (workarounds exist):**
-- 3 known limitations with acceptable workarounds
+- 2 known limitations with acceptable workarounds
 - 2 documentation TODOs
 - 1 dead code cleanup
 
-**Total Items:** 13
+**Total Items:** 12

src/client/dlq.rs

@@ -542,19 +542,22 @@ impl BackfillClient {
     pub async fn process_failed_jobs(&self) -> Result<u32, BackfillError> {
         // Find jobs that have failed permanently (attempts >= max_attempts)
         // and haven't been processed yet
-        // Note: The jobs view doesn't include payload, so we'll handle this limitation
         let find_failed_jobs_query = format!(
             r#"
-            SELECT id, task_identifier, queue_name, priority, key as job_key,
-                   max_attempts, attempts, last_error, created_at, run_at, updated_at
-            FROM {}.jobs
-            WHERE attempts >= max_attempts
-              AND max_attempts > 0
-              AND id NOT IN (SELECT COALESCE(original_job_id, -1) FROM {}.backfill_dlq)
-            ORDER BY updated_at ASC
+            SELECT jobs.id, tasks.identifier AS task_identifier,
+                   job_queues.queue_name, jobs.priority, jobs.key as job_key,
+                   jobs.max_attempts, jobs.attempts, jobs.last_error,
+                   jobs.created_at, jobs.run_at, jobs.updated_at, jobs.payload
+            FROM {}._private_jobs AS jobs
+            INNER JOIN {}._private_tasks AS tasks ON tasks.id = jobs.task_id
+            LEFT JOIN {}._private_job_queues AS job_queues ON job_queues.id = jobs.job_queue_id
+            WHERE jobs.attempts >= jobs.max_attempts
+              AND jobs.max_attempts > 0
+              AND jobs.id NOT IN (SELECT COALESCE(original_job_id, -1) FROM {}.backfill_dlq)
+            ORDER BY jobs.updated_at ASC
             LIMIT 100
         "#,
-            self.schema, self.schema
+            self.schema, self.schema, self.schema, self.schema
         );
 
         let failed_jobs = sqlx::query(&find_failed_jobs_query).fetch_all(&self.pool).await?;
@@ -564,8 +567,7 @@ impl BackfillClient {
         for job_row in failed_jobs {
             let job_id: i64 = job_row.get("id");
             let task_identifier: String = job_row.get("task_identifier");
-            // Payload is not available in the jobs view, use empty object as placeholder
-            let payload = serde_json::json!({});
+            let payload: serde_json::Value = job_row.get("payload");
             let queue_name: Option<String> = job_row.get("queue_name");
             let queue_name = queue_name.unwrap_or_else(|| "default".to_string());
             let priority: i16 = job_row.get("priority");

tests/dlq_tests.rs

@@ -204,6 +204,57 @@ async fn test_dlq_add_job_and_retrieve() {
     let _testjob: TestJob = serde_json::from_value(retrieved.payload).expect("the payload should be a test job");
 }
 
+#[tokio::test]
+async fn test_process_failed_jobs_preserves_payload() {
+    let client = setup_test_client("dlq_payload_preservation").await;
+    client.init_dlq().await.expect("DLQ init should work");
+
+    // Enqueue a job with specific payload
+    let test_payload = TestJob {
+        message: "unique test message for process_failed_jobs".to_string(),
+        number: 12345,
+    };
+
+    let outcome = client
+        .enqueue(
+            "test_job",
+            &test_payload,
+            JobSpec {
+                max_attempts: Some(1),
+                ..Default::default()
+            },
+        )
+        .await
+        .expect("should enqueue");
+
+    let job = outcome.unwrap();
+
+    // Simulate job failure by updating attempts to match max_attempts
+    sqlx::query(&format!(
+        "UPDATE {}._private_jobs SET attempts = max_attempts WHERE id = $1",
+        client.schema()
+    ))
+    .bind(job.id())
+    .execute(client.pool())
+    .await
+    .expect("should update job");
+
+    // Process failed jobs
+    let moved = client.process_failed_jobs().await.expect("should process");
+    assert_eq!(moved, 1);
+
+    // Verify payload was preserved
+    let dlq_jobs = client.list_dlq_jobs(DlqFilter::default()).await.expect("should list");
+    assert_eq!(dlq_jobs.jobs.len(), 1);
+
+    let dlq_job = &dlq_jobs.jobs[0];
+    let recovered: TestJob =
+        serde_json::from_value(dlq_job.payload.clone()).expect("payload should deserialize to TestJob");
+
+    assert_eq!(recovered.message, "unique test message for process_failed_jobs");
+    assert_eq!(recovered.number, 12345);
+}
+
 #[tokio::test]
 async fn test_dlq_list_with_filtering() {
     let client = setup_test_client("dlq_list_filter").await;