fix: preserve queue_name when adding jobs to DLQ

Author: ceejbot

docs/implementation/incomplete-features.md

@@ -124,30 +124,7 @@ QueueStatus {
 
 ## Low Priority - Known Limitations/Workarounds
 
-### 8. DLQ queue_name tracking
-**Location:** `src/client/dlq.rs:443-446`
-
-**Current State:** Uses hardcoded "default" queue name when adding jobs to DLQ
-
-**Issue:** `Job` struct from graphile_worker doesn't expose queue_name field
-
-**Code:**
-```rust
-// Note: Job doesn't have queue_name field, so we'll use "default" for now
-// In production, this would need to be tracked elsewhere or passed as a
-// parameter
-let queue_name = "default";
-```
-
-**Affects:** `add_to_dlq()` method
-
-**Possible Solutions:**
-- Add queue_name parameter to `add_to_dlq()`
-- Track queue_name in application state
-- Query GraphileWorker schema to get queue info
-- Check if newer graphile_worker versions expose this field
-
-### 9. DLQ pagination optimization
+### 8. DLQ pagination optimization
 **Location:** `src/client/dlq.rs:238`
 
 **Current State:** Uses separate COUNT query for pagination, could be more efficient
@@ -231,8 +208,8 @@ pub(crate) struct JobHandlerConfig {
 - All feature-gated, so not blocking main library use
 
 **Low Priority (workarounds exist):**
-- 2 known limitations with acceptable workarounds
+- 1 known limitation with acceptable workaround
 - 2 documentation TODOs
 - 1 dead code cleanup
 
-**Total Items:** 12
+**Total Items:** 11

src/client/dlq.rs

@@ -465,10 +465,20 @@ impl BackfillClient {
     ) -> Result<DlqJob, BackfillError> {
         let start = std::time::Instant::now();
 
-        // Note: Job doesn't have queue_name field, so we'll use "default" for now
-        // In production, this would need to be tracked elsewhere or passed as a
-        // parameter
-        let queue_name = "default";
+        // Query queue_name from job_queue_id
+        let queue_name = if let Some(queue_id) = original_job.job_queue_id() {
+            let query = format!(
+                "SELECT queue_name FROM {}._private_job_queues WHERE id = $1",
+                self.schema
+            );
+            sqlx::query_scalar::<_, String>(&query)
+                .bind(queue_id)
+                .fetch_optional(&self.pool)
+                .await?
+                .unwrap_or_else(|| "default".to_string())
+        } else {
+            "default".to_string()
+        };
 
         let insert_query = format!(
             r#"

tests/dlq_tests.rs

@@ -255,6 +255,41 @@ async fn test_process_failed_jobs_preserves_payload() {
     assert_eq!(recovered.number, 12345);
 }
 
+#[tokio::test]
+async fn test_add_to_dlq_preserves_queue_name() {
+    let client = setup_test_client("dlq_queue_name").await;
+    client.init_dlq().await.expect("DLQ init should work");
+
+    let test_job = TestJob {
+        message: "queue name test".to_string(),
+        number: 999,
+    };
+
+    // Enqueue to "fast" queue
+    let outcome = client
+        .enqueue(
+            "test_job",
+            &test_job,
+            JobSpec {
+                queue: Queue::Fast,
+                ..Default::default()
+            },
+        )
+        .await
+        .expect("should enqueue");
+
+    let job = outcome.unwrap();
+
+    // Add to DLQ
+    let dlq_job = client
+        .add_to_dlq(&job, "Test failure", None)
+        .await
+        .expect("should add to DLQ");
+
+    // Verify queue_name was preserved
+    assert_eq!(dlq_job.queue_name, "fast");
+}
+
 #[tokio::test]
 async fn test_dlq_list_with_filtering() {
     let client = setup_test_client("dlq_list_filter").await;