Fix DeferError handling when job re-enqueue fails

When a deferred job couldn't be re-enqueued due to concurrency limits,
two issues occurred:
1. The pk was left as None after delete() in the rolled-back transaction
2. DeferError was caught by generic Exception handler and logged at ERROR

This fix:
- Saves and restores self.pk before raising DeferError so convert_to_result works
- Adds dedicated DeferError handler with warning-level logging

Author: davegaeddert

plain-jobs/plain/jobs/models.py

@@ -265,6 +265,19 @@ def run(self) -> JobResult:
                 span.set_status(trace.StatusCode.OK)
                 return self.convert_to_result(status=JobResultStatuses.SUCCESSFUL)
 
+            except DeferError as e:
+                # Defer failed (e.g., concurrency limit reached during re-enqueue)
+                # The transaction was rolled back, so the JobProcess still exists in DB.
+                # The pk was restored in defer() before raising, so we can proceed normally.
+                logger.warning("Defer failed for %s: %s", self.job_class, e)
+                span.record_exception(e)
+                span.set_status(trace.Status(trace.StatusCode.ERROR, str(e)))
+                span.set_attribute(ERROR_TYPE, type(e).__name__)
+                return self.convert_to_result(
+                    status=JobResultStatuses.ERRORED,
+                    error=str(e),
+                )
+
             except Exception as e:
                 logger.exception(e)
                 span.record_exception(e)
@@ -294,7 +307,8 @@ def defer(self, *, job: Job, defer_exception: DeferJob) -> JobResult:
         )
 
         with transaction.atomic():
-            # 1. Save JobProcess UUID and delete (releases concurrency slot)
+            # 1. Save JobProcess state and delete (releases concurrency slot)
+            saved_pk = self.pk
             job_process_uuid = self.uuid
             job_request_uuid = self.job_request_uuid
             started_at = self.started_at
@@ -312,6 +326,8 @@ def defer(self, *, job: Job, defer_exception: DeferJob) -> JobResult:
 
             # Check if re-enqueue failed
             if new_job_request is None:
+                # Restore pk since transaction will roll back and object still exists
+                self.pk = saved_pk
                 raise DeferError(
                     f"Failed to re-enqueue deferred job {self.job_class}: "
                     f"concurrency limit reached for key '{self.concurrency_key}'"