fix(dlq): preserve parallel queue type when adding jobs to DLQ (#9)

## Summary

- Fix parallel jobs being converted to serial when requeued from DLQ
- Use empty string instead of "default" for parallel jobs in DLQ

## Problem

When parallel jobs (`Queue::Parallel`) fail and are added to the DLQ,
they were being stored with `queue_name = "default"`. When requeued, the
requeue logic sees a non-empty string and creates
`Queue::Serial("default")` instead of `Queue::Parallel`.

The requeue logic already correctly handles this distinction:
```rust
let queue = if dlq_job.queue_name.is_empty() {
    Queue::Parallel      // ← Empty string = parallel
} else {
    Queue::Serial(dlq_job.queue_name.clone())  // ← Non-empty = serial
};
```

But `add_to_dlq()` and `process_failed_jobs()` were defaulting to
`"default"` instead of empty string for parallel jobs (which have no
`job_queue_id`).

## Fix

- `add_to_dlq()`: Use `String::new()` when `job_queue_id()` is `None`
(parallel job)
- `process_failed_jobs()`: Use `unwrap_or_default()` when `queue_name`
is `NULL`

This ensures parallel jobs stay parallel when requeued from DLQ.

## Test plan

- [x] `cargo check` passes
- [x] `just fmt` - no changes needed

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-authored-by: Claude Opus 4.5 <noreply@anthropic.com>

Author: rrogers-machinify

src/client/dlq.rs

@@ -493,7 +493,9 @@ impl BackfillClient {
     ) -> Result<DlqJob, BackfillError> {
         let start = std::time::Instant::now();
 
-        // Query queue_name from job_queue_id
+        // Query queue_name from job_queue_id.
+        // For parallel jobs (no queue_id), use empty string so requeue preserves
+        // parallel execution. The requeue logic treats empty string as Queue::Parallel.
         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",
@@ -505,7 +507,9 @@ impl BackfillClient {
                 .await?
                 .unwrap_or_else(|| "default".to_string())
         } else {
-            "default".to_string()
+            // Parallel jobs have no queue_id - use empty string to preserve
+            // parallel execution when requeued (is_empty() check in requeue_dlq_job)
+            String::new()
         };
 
         // Use UPSERT to handle the case where a requeued job fails again.
@@ -618,7 +622,9 @@ impl BackfillClient {
             let task_identifier: String = job_row.get("task_identifier");
             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());
+            // Use empty string for parallel jobs (NULL queue_name) to preserve
+            // parallel execution when requeued
+            let queue_name = queue_name.unwrap_or_default();
             let priority: i16 = job_row.get("priority");
             let job_key: Option<String> = job_row.get("job_key");
             let max_attempts: i16 = job_row.get("max_attempts");

tests/dlq_tests.rs

@@ -290,6 +290,60 @@ async fn test_add_to_dlq_preserves_queue_name() {
     assert_eq!(dlq_job.queue_name, "fast");
 }
 
+#[tokio::test]
+async fn test_parallel_job_stays_parallel_through_dlq() {
+    let client = setup_test_client("dlq_parallel_roundtrip").await;
+    client.init_dlq().await.expect("DLQ init should work");
+
+    let test_job = TestJob {
+        message: "parallel job test".to_string(),
+        number: 777,
+    };
+
+    // Enqueue as parallel job (default)
+    let outcome = client
+        .enqueue(
+            "test_job",
+            &test_job,
+            JobSpec {
+                job_key: Some("parallel_dlq_test".to_string()),
+                ..Default::default() // Queue::Parallel is the default
+            },
+        )
+        .await
+        .expect("should enqueue");
+
+    let job = outcome.unwrap();
+
+    // Verify original job has no queue_id (parallel)
+    assert!(job.job_queue_id().is_none(), "parallel job should have no queue_id");
+
+    // Add to DLQ
+    let dlq_job = client
+        .add_to_dlq(&job, "Test failure", None)
+        .await
+        .expect("should add to DLQ");
+
+    // Verify queue_name is empty string (not "default")
+    assert!(
+        dlq_job.queue_name.is_empty(),
+        "parallel job should have empty queue_name in DLQ, got: '{}'",
+        dlq_job.queue_name
+    );
+
+    // Requeue the job
+    let requeued = client
+        .requeue_dlq_job(dlq_job.id, Some("Testing parallel preservation".to_string()))
+        .await
+        .expect("should requeue");
+
+    // Verify requeued job is still parallel (no queue_id)
+    assert!(
+        requeued.job_queue_id().is_none(),
+        "requeued parallel job should still have no queue_id"
+    );
+}
+
 #[tokio::test]
 async fn test_dlq_list_with_filtering() {
     let client = setup_test_client("dlq_list_filter").await;