streaQ v8.0.0

v8 is a large, breaking release with a new dead-letter queue, reworked task pipelining, and a refreshed web UI. It contains a variety of improvements including security, bugfixes, and performance.

⚠️ Upgrading from v7

Not only has streaQ's API changed, but also its core data format and Lua scripts. This means v8 workers cannot run tasks enqueued with v7 and vice versa. In fact, for safety v8 uses a modified queue prefix to ensure it won't even see your v7 queues (and vice versa). This makes it easy to migrate safely:

1. Move all enqueuing logic from v7 to v8.
2. Start new v8 workers to pick up new tasks.
3. Continue running v7 workers until the v7 queue is drained.

At this point you can safely delete v7 keys and stop all v7 workers.

Pipelining redesign

• then() and otherwise() are curried: Both now take the next task and return a function, which you call with that task's remaining arguments. This separates how tasks are linked from what each one receives. It lets type checkers verify the child's signature against the parent's output and allows positional arguments in addition to forwarded ones:

  # v7
  await my_fn.enqueue().then(next_fn, kwarg="a")
  
  # v8
  await my_fn.enqueue().then(next_fn)(kwarg="a")

• Chaining operators removed: | and ^ are gone: use then() and otherwise() instead. The operators were a precedence trap, as await a | b parses as (await a) | b, so every use needed extra parentheses, and since | and ^ have different precedence, a | b ^ c parses as a | (b ^ c) and silently builds the wrong chain.

  # v7
  await (fetch.enqueue(url) | parse | store)
  await (risky.enqueue(x) ^ safe)
  
  # v8
  await fetch.enqueue(url).then(parse)().then(store)()
  await risky.enqueue(x).otherwise(safe)()

• The parent's output is the child's first argument: Tuple outputs are no longer unpacked. Whatever the parent returns is passed as the child's first positional argument, and anything you pass when building the chain comes after it. So now typing is more straightforward and you get more flexibility to pass additional arguments if necessary.

  @worker.task
  async def tuplify(x: int) -> tuple[int, int]:
      return x, x * 2
  
  # v7: the tuple was spread across the child's parameters
  @worker.task
  async def untuple(a: int, b: int, third: int) -> int: ...
  await tuplify.enqueue(3).then(untuple, third=3)
  
  # v8: the child takes the tuple as a single argument
  @worker.task
  async def untuple(pair: tuple[int, int], third: int) -> int:
      a, b = pair
      ...
  await tuplify.enqueue(3).then(untuple)(3)

• forward=False runs a task after another without passing its output: then(task, forward=False) waits for the parent to succeed but doesn't pass its result on, so the child gets only the arguments you give it. Some chains are about ordering, not data: send a notification after something finishes, or clean up after a job. v7 had no way to express that in a chain without the child accepting, and then ignoring, the parent's output.

  await create_user.enqueue(args).then(notify, forward=False)(user_id)

• otherwise() can use its own arguments: By default, a fallback still replays the failed task's args and kwargs, as in v7. With forward=False, it uses the arguments you pass instead. Either way it's skipped and returns the parent's result if the parent succeeds.

  # v7: always replayed the parent's arguments
  await fetch.enqueue(url).otherwise(fetch_from_mirror)
  
  # v8: same behaviour...
  await fetch.enqueue(url).otherwise(fetch_from_mirror)()
  # ...or with different arguments
  await fetch.enqueue(url).otherwise(fetch_cached, forward=False)(url, max_age=3600)

• Built-in map() and filter(): v8 provides new built-in pipelining utilities for common operations:

  • task.map(fn)(...) runs fn on each item of task's output, in parallel, then aggregates the results into a list.
  • task.filter(predicate)(...) runs predicate on each item of task's output, in parallel, and keeps the items for which predicate is truthy.
  • fn.map(items)(...) and predicate.filter(items)(...) do the same, but start from an iterable directly.

  Extra arguments can be passed to every call and results come back in input order. If any subtask fails, the whole step fails, with the subtask's exception as the cause. The built-ins give subtasks deterministic IDs, so a retried map doesn't re-run items that already succeeded, and they use a single pub/sub connection to listen for results, regardless of how many items there are.

  # v7: the recipe from the docs
  await map.enqueue(data, to=double.fn_name).then(filter, by=is_even.fn_name)
  
  # v8
  await double.map(data)().filter(is_even)()
  await fetch_ids.enqueue().map(fetch_user)(timeout=5)  # extra args are duplicated per call

• start(after=...) removed (no more fan-in or DAG): Dependencies are expressed only through chains. Each task has at most one parent and one child, and a task can no longer wait on arbitrary task IDs. Since fan-in needed per-task dependency sets and extra Lua to keep them consistent, and it was the main source of leaked and orphaned tasks. Linear chains can be stored on the tasks themselves, with deterministic child IDs, and this simplification makes new features possible: idempotency and retrying pipelined tasks.

  So this code is no longer works directly:

  # task3 waits for both task1 and task2
  task1 = await step.enqueue(1)
  task2 = await step.enqueue(2)
  task3 = await later.enqueue().start(after=[task1.id, task2.id])

  However, this and other complex relationships can still be expressed via orchestrators.

• Only the first task in a chain can be configured: start() raises on any link except the first. Cron schedules and tasks registered with expire can't be chained either.

  await a.enqueue().then(b).start(priority="high")  # no longer allowed

New features

• Dead-letter queue. Set Worker(dlq_ttl=...) to park failed tasks instead of discarding them: 0 (default) disables it, None keeps them forever. Parked tasks can be inspected, requeued with requeue_by_id() or removed with purge_by_id(), from code or from the UI. Aborted, expired, and dependency-failed tasks aren't parked, nor are tasks with a fallback.
• New statuses: DEAD and DEPENDS. get_tasks_by_status(DONE) excludes dead tasks when the DLQ is on.
• Web UI: DEAD and DEPENDS views, requeue/purge/abort buttons on rows and task pages, priority filtering, loading indicators, compact counts (1.235M), and a wider layout.
• Stronger idempotency guarantees: guarantees last as long as task results, instead of until task finishes.

Smaller breaking API changes

• unique parameter removed: The per-task Redis lock REDIS_UNIQUE is gone from the task/cron decorators, RegisteredTask, finish_task, and the docs. Cron jobs no longer default to unique=True + 1h timeout, and the "unique tasks must have a timeout" error is gone. Uniqueness was a clunky solution to problems that no longer exist thanks to the atomic cron scheduling script present in streaQ for a while now.
• Removed streaq.utils.to_tuple()
• max_schedule_drift for cron tasks is now expire: a payload TTL measured from when the task becomes due. It isn't compatible with idempotency or the DLQ.
• The cron task timeout default changed from 1 hour to None.
• TaskStatus.NOT_FOUND is removed: status() simply returns None for missing tasks.
• TaskInfo.dependencies/dependents are replaced by child as multiple dependencies for a single task are no longer possible.

Bugfixes

• Fixed some small type lies in Lua refresh_timeout (now claim_task) and publish_task scripts.
• _previous_key no longer gets a hardcoded 5-min expiry.
• run_task checks the reclaim result before the expiry check, fixing a bug where an expired task error could be recorded for a task another worker owned.
• Stale abortion flag removed when the task already finished, which previously could persist indefinitely.
• Fixed #178
• Security: the web UI now escapes all output; task results, exception messages, and IDs could previously inject HTML/JS.
• A worker holding a stale prefetched message no longer counts a try or consumes an abort flag that belongs to the worker actually running the task.
• Children of tasks whose payload expired or couldn't be deserialized now fail instead of waiting forever.
• Passing the same priorities list to several workers no longer reverses it.
• Unscheduling a cron job mid-tick no longer crashes the worker.
• Invalid priorities raise in start(priority=...) instead of silently enqueuing to a stream that will never be read.

Other changes

• R TypeVar became covariant R_co everywhere.
• Security: Warn when pickle is used without a signing_secret, since Redis write access is sufficient for an attacker to achieve arbitrary code execution
• streaq web binds to 127.0.0.1 by default instead of 0.0.0.0

Pre-merge checklist

• Code formatted correctly (check with make lint)
• Passing tests locally (check with make test)
• New tests added (if applicable)
• Docs updated (if applicable)