Repository navigation
Bug: Retrying task produces the error as the pipeline result #26
Description
Description
- Schedule a pipeline and wait for its result.
- The first task in the pipeline raises an exception and goes to retry.
- Retrieve the pipeline result as an error.
- The pipeline then successfully completes with the correct result.
Expected
Await the pipeline’s result while a task is retrying.
Actual
The first exception is returned as the result regardless of retries.
Code
import asyncio
import math
from taskiq import Context
from taskiq import SimpleRetryMiddleware
from taskiq import TaskiqDepends
from taskiq import InMemoryBroker
from taskiq_pipelines import Pipeline
from taskiq_pipelines import PipelineMiddleware
from taskiq_redis import RedisAsyncResultBackend
from taskiq_redis import RedisStreamBroker
result_backend = RedisAsyncResultBackend(
redis_url="redis://localhost:6379",
)
broker = (
RedisStreamBroker(
url="redis://localhost:6379",
)
.with_middlewares(
PipelineMiddleware(),
SimpleRetryMiddleware(default_retry_count=3),
)
.with_result_backend(result_backend)
)
check_interval = 0.2
@broker.task("power", retry_on_error=True)
async def power(x: int, y: int = 2, context: "Context" = TaskiqDepends()) -> int:
if context.message.labels.get("_retries", 0) == 0:
raise ValueError()
await asyncio.sleep(check_interval * 10)
result = x**y
print(f"{x} ** {y} = {result}")
return result
@broker.task("sqrt", retry_on_error=True)
async def sqrt(value: int) -> float:
result = math.sqrt(value)
print(f"sqrt({value}) = {result}")
return result #
pipeline = Pipeline(broker).call_next(power).call_next(sqrt)
async def main():
task = await pipeline.kiq(2)
result = await task.wait_result(check_interval=check_interval)
print("result:", result)
if __name__ == "__main__":
asyncio.run(main())
Activity
harlanrs commentedon Dec 10, 2025
Potentially related details but let me know if I should raise this as its own issue.
I was running into a serialization issue when using both
PipelineMiddlewareandSmartRetryMiddlewarewithRedisStreamBrokerandListRedisScheduleSource.The pipeline would work fine until a task triggered a delayed retry which would be funneled through the scheduler instance.
Here's an example of the pipeline labels logged in the task before retry:
And here's how these labels were logged after the task was triggered by the scheduler:
Note that the
_pipe_datalabel was converted from a bytes json array to a base64 string and_pipe_current_stepwas converted from an int to a string.This would lead to a
JSONDecodeErrorin the post_save (line 44 of taskiq_pipeline/middleware.py) due to the pipeline label being a base64 string.I was able to bandaid the serialization error by adding a custom middleware first ahead of the other middlewares that specifically checked
_pipe_dataand base64 decoded it if needed. I am using theORJSONSerializerso I could simply check if the label was a string but this might require an alternative check when using the standard json serializer.I also went ahead and converted
_pipe_current_stepback to an int from a string to match the label format pre-scheduler but haven't specifically verified if this is required.With this custom middleware in place and 'correcting' the labels post-scheduler, the pipeline appears to be fully functional again.
Hope this provides some guidance or clues for others experiencing this issue.
cw-jlu commentedon Nov 12, 2026
I reproduced the premature final error on current master
d5d4da6e1733e2886ef955b505f7818690316587with Python 3.14.7, Taskiq 0.12.6 and the realInMemoryBrokerreceiver path. The first step fails once; its retry then waits on an asyncio Event. While that retry is still blocked,result_backend.is_result_ready(pipeline_task_id)is already true and the pipeline returns the first failure instead of waiting.PipelineMiddleware.on_error()publishes the final-step failure before Taskiq has completed all retry middleware decisions. This also happens when retry middleware suppresses that attempt's result withNoResultError.I plan to move terminal-error propagation to the post-save stage, after Taskiq has decided whether to save the attempt, with regressions for both middleware orders, successful retry, exhausted retries, and ordinary terminal failures. I will also check the result persistence against a real local Redis backend separately. Could a maintainer assign this to cw-jlu? I found no overlapping retry fix; the open group PR does not modify this middleware. AI-assisted work; the PR will remain Draft for independent and account-owner review.