René's URL Explorer Experiment

{"@context":"https://schema.org","@type":"QAPage","mainEntity":{"@type":"Question","name":"Can long-running tasks that haven't been completed be requeued after a graceful shutdown times out?","text":"
I have a test task that takes more than 30 seconds to execute as follows:
\n
@broker.task()\nasync def test(v: int):\n    print(f\"start {v}\")\n    await asyncio.sleep(35)\n    print(f\"end {v}\")\n
\n
I started TaskIQ with the following command: taskiq worker tasks.taskiq_init:broker -fsd --pattern=\"tasks/taskiq/tasks/*.py\" --shutdown-timeout=5 --wait-tasks-timeout=25.
\nWhen the test task began executing, I quickly used Ctrl+C and wait taskiq shutdown(the test task wasn't finished during the shutdown). Then I ran TaskIQ again. The test task did not re-execute. This behavior is different from sidekiq and Celery, which both requeue tasks that were not completed during shutdown.
\n
Is there an issue with the way I'm using TaskIQ(I use redis-broker)? Or does TaskIQ not have this feature? Is there any way to implement this functionality myself? This is a crucial feature for us because our project involves a large number of long-running tasks, each of which may take several minutes to execute.
\n
sincerely thanks!
","upvoteCount":1,"answerCount":1,"acceptedAnswer":{"@type":"Answer","text":"
I found a problem with redis streams. Suddenly whem you fetch new messages from the stream, redis doesn't send unacknowledged messages along with new ones. Clients should fetch and claim them themselves.
\n
I've updated the way we handle unacknowledged messages and released new version of taskiq-redis. https://github.com/taskiq-python/taskiq-redis/releases/tag/1.0.6 Please try it out.
","upvoteCount":0,"url":"https://github.com/orgs/taskiq-python/discussions/447#discussioncomment-12874089"}}}