Debugging a Production Error — Agent Found the Bug From Logs Alone

2026-03-11T15:47:23Z ERROR django.request Internal Server Error: /api/jobs/complete
Traceback (most recent call last):
  File "/app/jobs/views.py", line 142, in complete_job
    job.status = "completed"
    job.completed_at = timezone.now()
    job.save()
  File "/app/venv/lib/python3.12/django/db/models/base.py", line 822, in save
    self._perform_update(...)
django.db.utils.IntegrityError: duplicate key value violates unique constraint "jobs_job_completion_unique"
DETAIL: Key (job_id, status)=(4521, completed) already exists.

15:47:23.101 POST /api/jobs/complete job_id=4521 user=sarah_k
15:47:23.108 POST /api/jobs/complete job_id=4521 user=sarah_k

# BEFORE: No locking — vulnerable to concurrent writes
def complete_job(request, job_id):
    job = Job.objects.get(id=job_id)  # Both requests read same state
    if job.status != "in_progress":
        return Response({"error": "Job not in progress"}, status=400)

    job.status = "completed"
    job.completed_at = timezone.now()
    job.result_data = request.data.get("result")
    job.save()  # Second request explodes here

# AFTER: select_for_update() acquires a row-level lock
from django.db import transaction

def complete_job(request, job_id):
    with transaction.atomic():
        job = (
            Job.objects
            .select_for_update()  # Row-level lock — second request waits
            .get(id=job_id)
        )
        if job.status != "in_progress":
            return Response({"error": "Job not in progress"}, status=409)

        job.status = "completed"
        job.completed_at = timezone.now()
        job.result_data = request.data.get("result")
        job.save()

    return Response({"status": "completed"})

from concurrent.futures import ThreadPoolExecutor
from django.test import TestCase, TransactionTestCase

class TestJobCompletionRace(TransactionTestCase):
    def test_concurrent_completion_does_not_duplicate(self):
        job = Job.objects.create(status="in_progress")

        def complete_request():
            from django.test import RequestFactory
            factory = RequestFactory()
            request = factory.post(
                f"/api/jobs/{job.id}/complete",
                data={"result": "done"},
                content_type="application/json"
            )
            from jobs.views import complete_job
            return complete_job(request, job.id)

        with ThreadPoolExecutor(max_workers=2) as executor:
            futures = [executor.submit(complete_request) for _ in range(2)]
            results = [f.result() for f in futures]

        status_codes = [r.status_code for r in results]
        self.assertIn(200, status_codes)  # One succeeds
        self.assertIn(409, status_codes)  # One gets conflict
        self.assertEqual(Job.objects.filter(id=job.id, status="completed").count(), 1)