Add stress test harness for concurrency bugs

.github/workflows/test.yml

@@ -54,3 +54,45 @@ jobs:
       - name: Report coverage
         run: |
           uv run coverage report --show-missing --skip-covered --include 'steady_queue/**'
+
+  stress-test:
+    runs-on: ubuntu-latest
+
+    services:
+      postgres:
+        image: postgres:17.4-alpine
+        env:
+          POSTGRES_DB: app
+          POSTGRES_USER: steady_queue
+          POSTGRES_PASSWORD: steady_queue
+        ports:
+          - 5432:5432
+        options: >-
+          --health-cmd "pg_isready -U steady_queue -d postgres"
+          --health-interval 5s
+          --health-timeout 5s
+          --health-retries 10
+
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Set up Python
+        uses: actions/setup-python@v3
+        with:
+          python-version: "3.14"
+
+      - name: Install UV
+        run: |
+          python -m pip install uv==0.10.2
+
+      - name: Install dependencies
+        run: |
+          uv sync
+
+      - name: Create queue database
+        run: |
+          PGPASSWORD=steady_queue psql -h localhost -U steady_queue -d postgres -c "CREATE DATABASE queue OWNER steady_queue"
+
+      - name: Run stress test (smoke)
+        run: |
+          uv run python tests/stress_test.py --tasks 10 --workload light --timeout 60

AGENTS.md

@@ -0,0 +1,9 @@
+# Steady Queue Agent documentation
+
+This is a Python + Django port of the Solid Queue task processing library for Ruby on Rails. We strive to maintain 1-1 structure and feature parity, while conforming to Django's conventions.
+
+## Useful commands
+
+- Use `uv run python ...` to ensure the Python environment is loaded properly before running a command.
+- Run tests with `make test`
+- Lint with `make lint`

Makefile

@@ -10,7 +10,8 @@ steady_queue:
 
 .PHONY: lint
 lint:
-	uv run pre-commit run --all-files
+	uv run ruff check --fix
+	uv run ruff format
 
 .PHONY: force-kill
 force-kill:

tests/dummy/tasks.py

@@ -63,3 +63,52 @@ def dummy_recurring_task_2():
 @task()
 def limited_task_with_lambda_key(account_id: int, message: str = "hello"):
     print(f"limited task for account {account_id}: {message}")
+
+
+@task(queue_name="default")
+def stress_counter_task(job_id: int, workload: str = "none"):
+    """Increment execution counter for a job_id. Used by the stress test to detect duplicates.
+
+    workload controls how much simulated work each task performs:
+      - "none":   just the counter write (fast, tests queue machinery only)
+      - "light":  ~10-30ms sleep + ~1k hash rounds + counter write
+      - "medium": ~30-100ms sleep + ~5k hash rounds + counter write
+      - "heavy":  ~100-300ms sleep + ~20k hash rounds + counter write
+    """
+    import hashlib
+    import random
+    import time
+
+    # Workload profiles: (sleep_min_ms, sleep_max_ms, hash_rounds_min, hash_rounds_max)
+    profiles = {
+        "none": (0, 0, 0, 0),
+        "light": (10, 30, 500, 1500),
+        "medium": (30, 100, 3000, 8000),
+        "heavy": (100, 300, 15000, 25000),
+    }
+    sleep_min, sleep_max, rounds_min, rounds_max = profiles.get(
+        workload, profiles["none"]
+    )
+
+    # IO-bound: simulate an external API call / file write
+    if sleep_max > 0:
+        time.sleep(random.randint(sleep_min, sleep_max) / 1000.0)
+
+    # CPU-bound: iterative hashing
+    if rounds_max > 0:
+        data = f"stress-task-{job_id}".encode()
+        for _ in range(random.randint(rounds_min, rounds_max)):
+            data = hashlib.sha256(data).digest()
+
+    # DB write: always runs (this is our duplicate-execution detector)
+    from django.db import connections
+
+    conn = connections["default"]
+    with conn.cursor() as cursor:
+        cursor.execute(
+            """
+            INSERT INTO stress_test_counter (job_id, exec_count) VALUES (%s, 1)
+            ON CONFLICT (job_id) DO UPDATE SET exec_count = stress_test_counter.exec_count + 1
+            """,
+            [job_id],
+        )

tests/settings.py

@@ -141,6 +141,11 @@
             "level": "INFO",
             "propagate": False,
         },
+        "django.tasks": {
+            "handlers": ["console"],
+            "level": "WARNING",
+            "propagate": False,
+        },
         "steady_queue": {
             "handlers": ["console"],
             "level": "WARNING",