Use XREADGROUP MAXCOUNT/CLAIM when available

.github/workflows/python-app.yml

@@ -18,8 +18,14 @@ jobs:
           - "3.13"
           - "3.14"
           - "3.14t"
+        redis-version:
+          - "7.0"  # oldest version supported
+          - "8.0"
+          - "8.10"
     env:
       TERM: 'dumb'
+      PYTHON_VERSION: ${{ matrix.python-version }}
+      REDIS_VERSION: ${{ matrix.redis-version }}
     steps:
     - uses: actions/checkout@v6
       with:
@@ -40,6 +46,9 @@ jobs:
     - name: Type check with pyright
       run: uv run pyright streaq/ tests/ example.py
 
+    - name: Type check with mypy
+      run: uv run mypy --strict tests/ example.py
+
     - name: Test with pytest
       run: |
-        docker compose run --rm tests uv run --locked --all-extras --dev pytest -n auto --dist=loadgroup --cov=streaq tests/
+        docker compose run --rm -e COVERAGE_EXCLUDE=${{ matrix.redis-version == '8.10' && 'xreadgroup-no-max-count' || 'xreadgroup-max-count' }} tests uv run --locked --all-extras --dev pytest -n auto --cov=streaq tests/

Makefile

@@ -1,16 +1,20 @@
-.PHONY: install lint test docs
+.PHONY: install lint test docs cleanup
+
+REDIS_VERSION ?= latest
+PYTHON_VERSION ?= 3.11
+COVERAGE_EXCLUDE ?= xreadgroup-no-max-count
 
 install:
 	uv sync --all-extras
 
 lint:
-	uv run ruff check --select I --fix
+	uv run ruff check --fix streaq/ tests/ example.py
 	uv run ruff format streaq/ tests/
-	uv run ruff check streaq/ tests/ example.py
 	uv run pyright streaq/ tests/ example.py
+	uv run mypy --strict tests/ example.py
 
 test:
-	PYTHON_VERSION=3.11 docker compose run --rm tests uv run --locked --all-extras --dev pytest -n auto --cov=streaq tests/
+	PYTHON_VERSION=$(PYTHON_VERSION) REDIS_VERSION=$(REDIS_VERSION) docker compose run --rm -e COVERAGE_EXCLUDE=$(COVERAGE_EXCLUDE) tests uv run --locked --all-extras --dev pytest -n auto --cov=streaq tests/
 
 docs:
 	uv run -m sphinx -T -b html -d docs/_build/doctrees -D language=en docs/ docs/_build/

README.md

@@ -10,7 +10,7 @@ Fast, async, fully-typed distributed task queue via Redis streams
 
 ## Features
 
-- Up to [5x faster](https://github.com/tastyware/streaq/tree/master/benchmarks) than `arq`
+- Up to [14x faster](https://github.com/tastyware/streaq/tree/master/benchmarks) than `arq`
 - Fully typed
 - Comprehensive documentation
 - Support for delayed/scheduled tasks

docker-compose.yml

@@ -1,6 +1,6 @@
 services:
   redis-master:
-    image: redis:latest
+    image: redis:${REDIS_VERSION:-latest}
     hostname: redis-master
     command:
       [
@@ -19,7 +19,7 @@ services:
       retries: 15
 
   slave-1:
-    image: redis:latest
+    image: redis:${REDIS_VERSION:-latest}
     hostname: slave-1
     depends_on:
       - redis-master
@@ -43,7 +43,7 @@ services:
       retries: 15
 
   slave-2:
-    image: redis:latest
+    image: redis:${REDIS_VERSION:-latest}
     hostname: slave-2
     depends_on:
       - redis-master
@@ -67,7 +67,7 @@ services:
       retries: 15
 
   sentinel-1:
-    image: redis:latest
+    image: redis:${REDIS_VERSION:-latest}
     hostname: sentinel-1
     depends_on:
       - redis-master
@@ -86,7 +86,7 @@ services:
       retries: 15
 
   sentinel-2:
-    image: redis:latest
+    image: redis:${REDIS_VERSION:-latest}
     hostname: sentinel-2
     depends_on:
       - redis-master
@@ -105,7 +105,7 @@ services:
       retries: 15
 
   sentinel-3:
-    image: redis:latest
+    image: redis:${REDIS_VERSION:-latest}
     hostname: sentinel-3
     depends_on:
       - redis-master
@@ -124,7 +124,7 @@ services:
       retries: 15
 
   sentinel-init:
-    image: redis:latest
+    image: redis:${REDIS_VERSION:-latest}
     command: >
       sh -c '
         redis-cli -h sentinel-1 -p 26379 SENTINEL ckquorum mymaster &&
@@ -145,7 +145,7 @@ services:
         condition: service_healthy
 
   cluster-1:
-    image: redis:latest
+    image: redis:${REDIS_VERSION:-latest}
     hostname: cluster-1
     command: redis-server --port 7000 --cluster-enabled yes --appendonly yes --protected-mode no
     healthcheck:
@@ -155,7 +155,7 @@ services:
       retries: 15
 
   cluster-2:
-    image: redis:latest
+    image: redis:${REDIS_VERSION:-latest}
     hostname: cluster-2
     command: redis-server --port 7001 --cluster-enabled yes --appendonly yes --protected-mode no
     healthcheck:
@@ -165,7 +165,7 @@ services:
       retries: 15
 
   cluster-3:
-    image: redis:latest
+    image: redis:${REDIS_VERSION:-latest}
     hostname: cluster-3
     command: redis-server --port 7002 --cluster-enabled yes --appendonly yes --protected-mode no
     healthcheck:
@@ -175,7 +175,7 @@ services:
       retries: 15
 
   cluster-4:
-    image: redis:latest
+    image: redis:${REDIS_VERSION:-latest}
     hostname: cluster-4
     command: redis-server --port 7003 --cluster-enabled yes --appendonly yes --protected-mode no
     healthcheck:
@@ -185,7 +185,7 @@ services:
       retries: 15
 
   cluster-5:
-    image: redis:latest
+    image: redis:${REDIS_VERSION:-latest}
     hostname: cluster-5
     command: redis-server --port 7004 --cluster-enabled yes --appendonly yes --protected-mode no
     healthcheck:
@@ -195,7 +195,7 @@ services:
       retries: 15
 
   cluster-6:
-    image: redis:latest
+    image: redis:${REDIS_VERSION:-latest}
     hostname: cluster-6
     command: redis-server --port 7005 --cluster-enabled yes --appendonly yes --protected-mode no
     healthcheck:
@@ -205,7 +205,7 @@ services:
       retries: 15
 
   cluster-init:
-    image: redis:latest
+    image: redis:${REDIS_VERSION:-latest}
     command: >
         sh -c '
           if redis-cli -h cluster-1 -p 7000 cluster info | grep -q "cluster_state:ok"; then

docs/task.rst

@@ -359,22 +359,23 @@ This is useful for ETL pipelines or similar tasks, where each task builds upon t
 .. code-block:: python
 
    from typing import Any
-   from streaq.utils import gather, to_tuple
+   from anyio import gather
+   from streaq.utils import to_tuple
 
    @worker.task
    async def map(data: list[Any], *, to: str) -> list[Any]:
-       task = worker.registry[to]
-       coros = [task.enqueue(*to_tuple(d)).start() for d in data]
-       tasks = await gather(*coros)
+       task = map.worker.registry[to]
+       tasks = [task.enqueue(*to_tuple(d)) for d in data]
+       await map.worker.enqueue_many(tasks)
        results = await gather(*[t.result(3) for t in tasks])
        return [r.result for r in results]
 
    @worker.task
    async def filter(data: list[Any], *, by: str) -> list[Any]:
-       task = worker.registry[by]
-       coros = [task.enqueue(*to_tuple(d)).start() for d in data]
-       tasks = await gather(*coros)
-       results = await gather(*[t.result(5) for t in tasks])
+       task = filter.worker.registry[by]
+       tasks = [task.enqueue(*to_tuple(d)) for d in data]
+       await filter.worker.enqueue_many(tasks)
+       results = await gather(*[t.result(3) for t in tasks])
        return [data[i] for i in range(len(data)) if results[i].result]
 
    async with worker:

pyproject.toml

@@ -54,8 +54,8 @@ authors = [
     { name = "Graeme Holliday", email = "graeme@tastyware.dev" }
 ]
 dependencies = [
-    "anyio>=4.13.0",
-    "coredis>=6.7.0",
+    "anyio>=4.15.0",
+    "coredis>=6.9.0",
     "crontab>=1.0.5",
     "typer>=0.19.2",
     "watchfiles>=1.1.0",
@@ -84,12 +84,13 @@ Changelog = "https://github.com/tastyware/streaq/releases"
 [dependency-groups]
 dev = [
     "httpx>=0.28.1",
-    "pyright>=1.1.406",
+    "mypy>=2.3.1",
+    "pyright>=1.1.411",
     "pytest>=8.4.2",
     "pytest-cov>=7.0.0",
     "pytest-randomly>=4.1.0",
     "pytest-xdist>=3.8.0",
-    "ruff>=0.13.1",
+    "ruff>=0.16.0",
     "sphinx>=8.1.3",
     "sphinx-immaterial>=0.13.6",
     "trio>=0.30.0",
@@ -100,10 +101,23 @@ dev = [
 testpaths = "tests"
 
 [tool.ruff.lint]
-select = ["E", "F", "I", "UP"]
+extend-select = ["ANN", "FAST"]
+extend-ignore = ["ANN401"]
+
+[tool.ruff.lint.per-file-ignores]
+"tests/**/*.py" = ["DTZ", "ANN201", "TRY002", "ASYNC"]
+
+[tool.ruff.format]
+docstring-code-line-length = 88
 
 [tool.pyright]
-strict = ["streaq/"]
+typeCheckingMode = "strict"
+reportPrivateUsage = "none"
+reportUnusedFunction = "none"
+
+[[tool.mypy.overrides]]
+module = ["tests.*"]
+disable_error_code = ["no-untyped-def"]
 
 [tool.coverage.run]
 source = ["streaq"]
@@ -114,6 +128,7 @@ patch = ["subprocess"]
 [tool.coverage.report]
 show_missing = true
 fail_under = 100
+exclude_also = ['# pragma: gated cover\[${COVERAGE_EXCLUDE?}\]']
 
 [tool.coverage.paths]
 streaq = ["streaq", "/app/streaq"]

streaq/__init__.py

@@ -1,13 +1,11 @@
 import logging
 
-VERSION = "7.1.0"
+VERSION = "7.2.0"
 __version__ = VERSION
 
 logger = logging.getLogger(__name__)
 logger.addHandler(logging.NullHandler())
 
-# ruff: noqa: E402
-
 from .task import TaskStatus
 from .types import StreaqError, StreaqRetry, TaskContext
 from .worker import Worker

streaq/lua/streaq.lua

@@ -59,19 +59,13 @@ redis.register_function('publish_delayed_tasks', function(keys, argv)
 
   local current_time = argv[1]
 
-  for i = 2, #argv do
-    local priority = argv[i]
-    local queue = queue_key .. priority
-    -- get and delete tasks ready to run from delayed queue (with scores)
-    local tids = redis.call('zrange', queue, 0, current_time, 'byscore', 'withscores')
-    if #tids > 0 then
-      redis.call('zremrangebyscore', queue, 0, current_time)
-
-      local stream = stream_key .. priority
-      -- add ready tasks to live queue, using scheduled fire time as enqueue_time
-      for j = 1, #tids, 2 do
-        redis.call('xadd', stream, '*', 'task_id', tids[j], 'enqueue_time', tids[j + 1])
-      end
+  -- get and delete tasks ready to run from delayed queue (with scores)
+  local tids = redis.call('zrange', queue_key, 0, current_time, 'byscore', 'withscores')
+  if #tids > 0 then
+    redis.call('zremrangebyscore', queue_key, 0, current_time)
+    -- add ready tasks to live queue, using scheduled fire time as enqueue_time
+    for j = 1, #tids, 2 do
+      redis.call('xadd', stream_key, '*', 'task_id', tids[j], 'enqueue_time', tids[j + 1])
     end
   end
 end)
@@ -86,10 +80,9 @@ redis.register_function('publish_task', function(keys, argv)
 
   local task_id = argv[1]
   local task_data = argv[2]
-  local priority = argv[3]
-  local score = argv[4]
-  local expire = argv[5]
-  local current_time = argv[6]
+  local score = argv[3]
+  local expire = argv[4]
+  local current_time = argv[5]
 
   local args
   if expire ~= '0' then
@@ -102,7 +95,7 @@ redis.register_function('publish_task', function(keys, argv)
 
   local modified = 0
   -- additional args are dependencies for task
-  for i = 7, #argv do
+  for i = 6, #argv do
     local dep_id = argv[i]
     -- update dependency DAG if dependency exists
     if redis.call('exists', results_key .. dep_id) ~= 1 then
@@ -116,10 +109,10 @@ redis.register_function('publish_task', function(keys, argv)
   if modified == 0 then
     -- delayed queue
     if score ~= '0' then
-      redis.call('zadd', queue_key .. priority, score, task_id)
+      redis.call('zadd', queue_key, score, task_id)
       -- live queue
     else
-      return redis.call('xadd', stream_key .. priority, '*', 'task_id', task_id, 'enqueue_time', current_time)
+      return redis.call('xadd', stream_key, '*', 'task_id', task_id, 'enqueue_time', current_time)
     end
   end