Add durable Redis backend

.github/workflows/ci.yml

@@ -27,6 +27,7 @@ jobs:
  - "3.14"
  django-version:
  - "6.1a1"
+ runs-on: ${{ matrix.os }}
  services:
  redis:
  image: redis
@@ -35,7 +36,6 @@ jobs:
  options: --entrypoint redis-server
  env:
  REDIS_URL: redis:///0
- runs-on: ${{ matrix.os }}
  steps:
  - uses: actions/checkout@v7
  - uses: astral-sh/setup-uv@v7
@@ -45,26 +45,3 @@ jobs:
  - uses: codecov/codecov-action@v7
  with:
  token: ${{ secrets.CODECOV_TOKEN }}
- pytest-windows-macos:
- name: Pytest
- permissions:
- contents: read
- strategy:
- matrix:
- os:
- - "windows-latest"
- - "macos-latest"
- python-version:
- - "3.13"
- django-version:
- - "6.1a1"
- runs-on: ${{ matrix.os }}
- steps:
- - uses: actions/checkout@v7
- - uses: astral-sh/setup-uv@v7
- with:
- python-version: ${{ matrix.python-version }}
- - uses: codecov/codecov-action@v7
- with:
- token: ${{ secrets.CODECOV_TOKEN }}
- - run: uv run --with django~=${{ matrix.django-version }} pytest -m "not benchmark"

CONTRIBUTING.md

@@ -9,8 +9,8 @@ curl -sSL https://raw.githubusercontent.com/codingjoe/naming-things/refs/heads/m
 
 ## Design Principles
 
-- Consistency – We never lose data, even if someone unplugs the power or network.
 - Durability – We recover from any failures, even poorly written tasks.
+- Consistency – We never lose data, even if someone unplugs the power or network.
 - Utilization – We keep the CPU saturated with tasks, not with idle time or waiting for locks.
 
 ## Testing

README.md

@@ -4,7 +4,7 @@
  <picture>
  <source media="(prefers-color-scheme: dark)" srcset="https://github.com/codingjoe/threadmill/raw/main/images/logo-dark.svg">
  <source media="(prefers-color-scheme: light)" srcset="https://github.com/codingjoe/threadmill/raw/main/images/logo-light.svg">
- <img alt="Django Grinder: A queue agnostic worker for Django's task framework." src="https://github.com/codingjoe/threadmill/raw/main/images/logo-light.svg">
+ <img alt="Threadmill: A queue agnostic worker for Django's task framework." src="https://github.com/codingjoe/threadmill/raw/main/images/logo-light.svg">
  </picture>
 <br>
  <a href="https://github.com/codingjoe/threadmill/">Documentation</a> |
@@ -34,20 +34,31 @@
 
 ## Setup
 
-You need to have [Django's Task framework][django-tasks] setup properly.
+You need to have [Django's Task framework][django-tasks] set up properly.
 
 ```console
-uv add threadmill
+uv add threadmill[redis]
 ```
 
-Add `threadmill` to your `INSTALLED_APPS` in `settings.py`:
+Add `threadmill` to your `INSTALLED_APPS` in `settings.py`
+and configure the task backend:
 
 ```python
 # settings.py
+import os
+
 INSTALLED_APPS = [
  "threadmill",
  # ...
 ]
+
+TASKS = {
+ "default": {
+ "BACKEND": "threadmill.backends.redis.RedisTaskBackend",
+ "REDIS_URL": os.getenv("REDIS_URL", "redis://localhost:6379/0"),
+ },
+ # ...
+}
 ```
 
 Finally, you launch the worker pool:
@@ -58,9 +69,11 @@ uv run manage.py threadmill
 
 ## Usage
 
+### Workers
+
 The workers are inspired by Gunicorn, and the CLI is very similar.
 
-### Utilization
+#### Utilization
 
 Depending on your workload, you can tweak the number of processes and threads.
 Processes allow for parallel compute (no GIL) while threads are great for low-memory concurrent IO.
@@ -69,7 +82,7 @@ Processes allow for parallel compute (no GIL) while threads are great for low-me
 uv run manage.py threadmill --processes 4 --threads 2
 ```
 
-### Health
+#### Health
 
 If your tasks leak memory, you can recycle (restart) the workers after a certain number of tasks have been processed:
 
@@ -81,61 +94,33 @@ This will restart the workers after 1000 tasks have been processed, with a rando
 
 Should a worker crash or be killed, the pool will automatically restart it.
 
-### Shutdown
+#### Shutdown
 
 A graceful shutdown is possible with the `SIGTERM` or a keyboard interrupt.
-All workers will finish the tasks they acquired and publish them.
+All workers will finish the tasks they acquired and acknowledge them.
 
 You can use `--exit-empty` to exit immediately after all tasks have been processed,
 which might be useful for draining a one-off queue.
 
-### Task Backlog
-
-You can prefetch tasks from a queue to avoid IO latency bottlenecks.
-However, this will increase the memory usage of the worker pool.
-
-```console
-uv run manage.py threadmill --prefetch 100
-```
-
-### Task Timeouts
-
-> [!WARNING]
-> Work in progress, this feature is not yet stable.
+### Redis Backend Options
 
-Task timeouts are important to ensure the long-term health of your pool.
-However, they need to be aligned with your queueing system's timeout settings.
-The message queue needs to requeue a task that hasn't been acknowledged within the timeout.
+The `RedisTaskBackend` accepts the following options under `OPTIONS` in your
+`TASKS` configuration:
 
-## Integration
+| Option | Default | Description |
+| ----------------- | ---------------------- | ------------------------------------------------------------ |
+| `lease_ttl` | `timedelta(hours=1)` | Max processing time before a started task is marked FAILED. |
+| `result_ttl` | `timedelta(days=1)` | How long task results are retained before automatic removal. |
+| `broker_interval` | `timedelta(seconds=1)` | Interval between background broker maintenance passes. |
+| `batch_size` | `100` | Max tasks to move or requeue per broker pass. |
 
-> [!NOTE]
-> This section is for people who want to integrate Threadmill into their queueing system.
+A task that is started but never acknowledged (lease expired) is marked FAILED
+with an `AcknowledgementTimeout` error. Set `lease_ttl` comfortably above your
+worst-case task runtime.
 
-Threadmill is designed to be durable and requires a queueing system to support late acknowledgement.
+All keys for one backend alias share a Redis Cluster hash tag (`{alias}`), so
+every multi-key operation — including the cross-queue acquire — runs on a single
+shard. Scale horizontally by running additional backend aliases, not by relying
+on cross-slot operations.
 
-To use Threadmill, your backend will need to inherit from `threadmill.backends.AcknowledgeableTaskBackend` and implement the following methods:
-
-```python
-class AcknowledgeableTaskBackend(BaseTaskBackend, ABC):
- """Provide an interface for tasks queues to be processed by the executor."""
-
- def acquire(
- self, *queue_names: str, timeout: datetime.timedelta | None = None
- ) -> TaskResult:
- """
- Return and lock the next task to be processed without removing it from the queue.
-
- Args:
- queue_names: The names of the queues to acquire tasks from.
- timeout: The maximum time to wait for a task. If None, wait indefinitely.
-
- Raises:
- TimeoutError: If no task is available within the specified timeout.
- """
- raise NotImplementedError
-
- def acknowledge(self, task_result: TaskResult) -> None:
- """Remove the task from the queue and publish the result."""
- raise NotImplementedError
-```
+[django-tasks]: https://docs.djangoproject.com/en/stable/topics/tasks/

pyproject.toml

@@ -35,6 +35,9 @@ classifiers = [
 requires-python = ">=3.12"
 dependencies = ["django>=6.1a1"]
 
+[project.optional-dependencies]
+redis = ["redis>=5.0"]
+
 [project.urls]
 # https://packaging.python.org/en/latest/specifications/well-known-project-urls/#well-known-labels
 Homepage = "https://github.com/codingjoe/threadmill"
@@ -56,9 +59,9 @@ minversion = "6.0"
 addopts = "--cov --cov-report=xml --cov-report=term --tb=short -rxs --benchmark-autosave --benchmark-group-by=fullname --benchmark-min-rounds=10"
 testpaths = ["tests"]
 DJANGO_SETTINGS_MODULE = "tests.testapp.settings"
+asyncio_mode = "auto"
 markers = [
  "benchmark: mark benchmark tests.",
- "integration: mark integration tests.",
 ]
 
 [tool.coverage.run]
@@ -91,6 +94,7 @@ combine-as-imports = true
 split-on-trailing-comma = true
 section-order = ["future", "standard-library", "third-party", "first-party", "local-folder"]
 force-wrap-aliases = true
+known-first-party = ["threadmill", "tests"]
 
 [tool.ruff.lint.pydocstyle]
 convention = "pep257"
@@ -105,4 +109,5 @@ test = [
  "pytest-asyncio",
  "pytest-cov",
  "pytest-django",
+ "redis>=5.0",
 ]

tests/backends/test_base.py

@@ -0,0 +1,116 @@
+from __future__ import annotations
+
+import datetime
+import time
+import uuid
+
+import pytest
+from django.tasks import TaskResult, TaskResultStatus
+from django.tasks.base import TaskError
+from django.utils import timezone
+
+from threadmill.backends.base import Broker, ThreadmillTaskBackend
+from threadmill.exceptions import AcknowledgementTimeout
+
+
+class BackendDouble(ThreadmillTaskBackend):
+ def enqueue(self, task, args, kwargs):
+ return TaskResult(
+ task=task,
+ id=str(uuid.uuid4()),
+ status=TaskResultStatus.READY,
+ enqueued_at=timezone.now(),
+ started_at=None,
+ finished_at=None,
+ last_attempted_at=None,
+ backend=self.alias,
+ errors=[],
+ worker_ids=[],
+ args=args,
+ kwargs=kwargs,
+ )
+
+
+class TestAcknowledgeableTaskBackend:
+ def test_acquire__raise_not_implemented_error(self) -> None:
+ """Raise NotImplementedError for backend acquire API."""
+ with pytest.raises(NotImplementedError):
+ BackendDouble(alias="default", params={}).acquire(
+ timeout=datetime.timedelta(seconds=1)
+ )
+
+ def test_acknowledge__raise_not_implemented_error(self) -> None:
+ """Raise NotImplementedError for backend acknowledge API."""
+ with pytest.raises(NotImplementedError):
+ BackendDouble(alias="default", params={}).acknowledge(task_result=None)
+
+ def test_peek__raise_not_implemented_error(self) -> None:
+ """Raise NotImplementedError for backend peek_results API."""
+ with pytest.raises(NotImplementedError):
+ list(BackendDouble(alias="default", params={}).peek("default"))
+
+
+class TestAcknowledgementTimeout:
+ """Tests for the AcknowledgementTimeout exception."""
+
+ def test_exception_can_be_instantiated(self) -> None:
+ """AcknowledgementTimeout can be instantiated."""
+ exc = AcknowledgementTimeout()
+ assert isinstance(exc, Exception)
+
+ def test_exception_can_be_used_in_task_error(self) -> None:
+ """AcknowledgementTimeout can be used as a TaskError's exception_class_path."""
+ error = TaskError(
+ exception_class_path="threadmill.exceptions.AcknowledgementTimeout",
+ traceback="Task processing lease expired.",
+ )
+ assert (
+ error.exception_class_path == "threadmill.exceptions.AcknowledgementTimeout"
+ )
+
+
+class FakeBroker(Broker):
+ """Broker that records main() calls for testing."""
+
+ def __init__(
+ self, *, interval: datetime.timedelta = datetime.timedelta(seconds=0.01)
+ ) -> None:
+ super().__init__(interval=interval)
+ self.maintain_calls: list[float] = []
+
+ def main(self) -> None:
+ self.maintain_calls.append(time.monotonic())
+
+
+class TestBroker:
+ def test_main__is_noop(self) -> None:
+ """Base Broker.main() is a no-op."""
+ Broker(interval=datetime.timedelta(seconds=1)).main()
+
+ def test_run__calls_maintain_then_exits_on_shutdown(self) -> None:
+ """run() loops calling main() and exits after shutdown()."""
+ broker = FakeBroker(interval=datetime.timedelta(seconds=0.01))
+ broker.start()
+ time.sleep(0.05)
+ broker.shutdown()
+ broker.join(timeout=1)
+ assert not broker.is_alive()
+ assert len(broker.maintain_calls) >= 1
+
+ def test_interval_is_honored(self) -> None:
+ """Broker waits at least interval between main() calls."""
+ broker = FakeBroker(interval=datetime.timedelta(seconds=0.1))
+ broker.start()
+ time.sleep(0.25)
+ broker.shutdown()
+ broker.join(timeout=1)
+ assert len(broker.maintain_calls) >= 2
+ for i in range(1, len(broker.maintain_calls)):
+ assert broker.maintain_calls[i] - broker.maintain_calls[i - 1] >= 0.09
+
+ def test_shutdown__sets_event(self) -> None:
+ """shutdown() sets the shutdown_requested event."""
+ broker = Broker(interval=datetime.timedelta(seconds=1))
+ assert not broker.shutdown_requested.is_set()
+ broker.shutdown()
+ assert broker.shutdown_requested.is_set()