Breaking changes for v7.0.0

.github/workflows/python-app.yml

@@ -13,7 +13,6 @@ jobs:
       fail-fast: false
       matrix:
         python-version:
-          - "3.10"
           - "3.11"
           - "3.12"
           - "3.13"

.python-version

@@ -1 +1 @@
-3.10
+3.11

Makefile

@@ -10,7 +10,7 @@ lint:
 	uv run pyright streaq/ tests/ example.py
 
 test:
-	PYTHON_VERSION=3.10 docker compose run --rm tests uv run --locked --all-extras --dev pytest -n auto --cov=streaq tests/
+	PYTHON_VERSION=3.11 docker compose run --rm 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/

benchmarks/README.md

@@ -1,11 +1,9 @@
 # Benchmarks
 
-streaQ's performance significantly improves upon [arq](https://github.com/python-arq/arq), and is on-par with [SAQ](https://github.com/tobymao/saq) and [taskiq](https://github.com/taskiq-python/taskiq). If you want to run these tests yourself, first install the dependencies:
+streaQ's performance significantly improves upon [arq](https://github.com/python-arq/arq), and is on-par with [taskiq](https://github.com/taskiq-python/taskiq). If you want to run these tests yourself, first install the dependencies:
 ```
 $ uv pip install git+https://github.com/Graeme22/arq.git
-$ uv pip install "saq[hiredis]==0.25.2"
-$ uv pip install "taskiq-redis==1.1.0"
-$ uv pip install uvloop
+$ uv pip install "taskiq-redis==1.2.2"
 ```
 
 You can enqueue jobs like so:
@@ -16,33 +14,30 @@ $ python benchmarks/bench_streaq.py --time 1
 And run a worker with one of these commands, adjusting the number of workers as desired:
 ```
 $ arq --workers ? --burst bench_arq.WorkerSettings
-$ saq --quiet bench_saq.settings --workers ?
 $ streaq run --burst --workers ? bench_streaq:worker
 $ taskiq worker --workers ? --max-async-tasks 32 bench_taskiq:broker --max-prefetch 32
 ```
 
-These benchmarks were run with streaQ v6.0.0 on an M4 Mac Mini using asyncio + uvloop. Trio performance is slightly worse.
+These benchmarks were run with streaQ v7.0.0 on an M4 Mac Mini using asyncio + uvloop. Trio performance is slightly worse.
 
 ## Benchmark 1: No-op
 
 This benchmark evaluates the performance when tasks do nothing, representing negligible amounts of work.
-These results are with 20,000 tasks enqueued, a concurrency of `32`, and a variable number of workers.
+These results are with 20,000 tasks enqueued, concurrency of `32`, and a variable number of workers.
 
-| library  | enqueuing | 1 worker | 10 workers | 20 workers | 40 workers |
-| -------- | --------- | -------- | ---------- | ---------- | ---------- |
-| streaq   | 0.45s     | 8.61s    | 3.90s      | 4.18s      | 4.84s      |
-| SAQ      | 1.67s     | 9.86s    | 3.46s      | 3.45s      | 3.93s      |
-| taskiq   | 1.68s     | 6.36s    | 3.26s      | 3.38s      | 6.43s      |
-| arq      | 2.31s     | 62.66s   | 28.10s     | 43.33s     | ☠️         |
+| library  | enqueuing | 1 worker | 10 workers |
+| -------- | --------- | -------- | ---------- |
+| streaq   | 0.57s     | 5.54s    | 2.45s      |
+| taskiq   | 1.60s     | 4.60s    | 3.00s      |
+| arq      | 1.32s     | 62.06s   | 35.88s     |
 
 ## Benchmark 2: Sleep
 
 This benchmark evaluates the performance when tasks sleep for 1 second, representing a small amount of work.
-These results are with 20,000 tasks enqueued, a concurrency of `32`, and a variable number of workers.
-
-| library  | enqueuing | 10 workers | 20 workers | 40 workers |
-| -------- | --------- | ---------- | ---------- | ---------- |
-| streaq   | 0.44s     | 64.56s     | 32.80s     | 17.88s     |
-| SAQ      | 1.69s     | 64.51s     | 33.56s     | 17.74s     |
-| taskiq   | 1.68s     | 67.53s     | 34.42s     | 18.55s     |
-| arq      | 2.27s     | 176.87s    | 169.47s    | ☠️         |
+These results are with 20,000 tasks enqueued, concurrency of `32`, and a variable number of workers.
+
+| library  | enqueuing | 10 workers | 100 workers |
+| -------- | --------- | ---------- | ----------- |
+| streaq   | 0.57s     | 64.20s     | 7.90s       |
+| taskiq   | 1.60s     | 63.57s     | 12.19s      |
+| arq      | 1.32s     | 178.96s    | 285.99s     |

benchmarks/bench_arq.py

@@ -21,11 +21,21 @@ async def sleeper(ctx, time: int) -> None:
         await asyncio.sleep(time)
 
 
+async def startup(ctx):
+    ctx["start_time"] = asyncio.get_event_loop().time()
+
+
+async def shutdown(ctx):
+    print(f"finished after {asyncio.get_event_loop().time() - ctx['start_time']}s")
+
+
 class WorkerSettings:
     functions = [sleeper]
     redis_settings = settings
     max_jobs = 32
     burst = True
+    on_startup = startup
+    on_shutdown = shutdown
 
 
 async def main(time: int):

docs/middleware.rst

@@ -10,7 +10,6 @@ You can define middleware to wrap task execution. This has a host of potential a
 
    import time
    from typing import Any
-   from streaq import TaskContext, TaskDepends
    from streaq.types import ReturnCoroutine
 
    @worker.middleware

pyproject.toml

@@ -37,7 +37,6 @@ classifiers = [
     "Programming Language :: Python",
     "Programming Language :: Python :: 3",
     "Programming Language :: Python :: 3 :: Only",
-    "Programming Language :: Python :: 3.10",
     "Programming Language :: Python :: 3.11",
     "Programming Language :: Python :: 3.12",
     "Programming Language :: Python :: 3.13",
@@ -49,14 +48,14 @@ classifiers = [
     "Topic :: System :: Systems Administration",
     "Typing :: Typed",
 ]
-requires-python = ">=3.10"
+requires-python = ">=3.11"
 license = {file = "LICENSE"}
 authors = [
     { name = "Graeme Holliday", email = "graeme@tastyware.dev" }
 ]
 dependencies = [
     "anyio>=4.13.0",
-    "coredis>=6.6.1",
+    "coredis>=6.7.0",
     "crontab>=1.0.5",
     "typer>=0.19.2",
     "watchfiles>=1.1.0",
@@ -88,6 +87,7 @@ dev = [
     "pyright>=1.1.406",
     "pytest>=8.4.2",
     "pytest-cov>=7.0.0",
+    "pytest-randomly>=4.1.0",
     "pytest-xdist>=3.8.0",
     "ruff>=0.13.1",
     "sphinx>=8.1.3",

streaq/__init__.py

@@ -1,6 +1,6 @@
 import logging
 
-VERSION = "6.5.3"
+VERSION = "7.0.0"
 __version__ = VERSION
 
 logger = logging.getLogger(__name__)
@@ -9,21 +9,7 @@
 # ruff: noqa: E402
 
 from .task import TaskStatus
-from .types import (
-    StreaqError,
-    StreaqRetry,
-    TaskContext,
-    TaskDepends,  # pyright: ignore[reportDeprecated]
-    WorkerDepends,  # pyright: ignore[reportDeprecated]
-)
+from .types import StreaqError, StreaqRetry, TaskContext
 from .worker import Worker
 
-__all__ = [
-    "StreaqError",
-    "StreaqRetry",
-    "TaskContext",
-    "TaskDepends",
-    "TaskStatus",
-    "Worker",
-    "WorkerDepends",
-]
+__all__ = ["StreaqError", "StreaqRetry", "TaskContext", "TaskStatus", "Worker"]

streaq/lua/streaq.lua

@@ -193,13 +193,16 @@ end)
 
 redis.register_function('refresh_timeout', function(keys, argv)
   local stream_key = keys[1]
+  local running_set = keys[2]
 
   local group_name = argv[1]
   local consumer = argv[2]
   local message_id = argv[3]
+  local task_id = argv[4]
 
   if #redis.call('xpending', stream_key, group_name, message_id, message_id, 1, consumer) > 0 then
     redis.call('xclaim', stream_key, group_name, consumer, 0, message_id, 'justid')
+    redis.call('sadd', running_set, task_id)
     return true
   end
   return false