Can long-running tasks that haven't been completed be requeued after a graceful shutdown times out?

[sujh]

I have a test task that takes more than 30 seconds to execute as follows:

@broker.task()
async def test(v: int):
    print(f"start {v}")
    await asyncio.sleep(35)
    print(f"end {v}")

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.
When 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.

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.

sincerely thanks!

[s3rius]

Can you share what broker do you use? And how do you configure it? Different brokers works differently with acknowledgements. Some of them don't support them.

[sujh]

@s3rius thanks for your reply.

we use RedisStreamBroker(1.0.5), it should support acknowledgement, but the acknowledgement mechanism seems not work. The tasks are easily lost when shutdown.

here is the configuration:

_redis_url = f"redis://:{settings.secret.REDIS_PASSWORD}@{settings.secret.REDIS_HOST}:{settings.secret.REDIS_PORT}/{settings.REDIS_DB}"

result_backend = RedisAsyncResultBackend(redis_url=_redis_url)
broker = RedisStreamBroker(url=_redis_url).with_result_backend(result_backend)

[s3rius]

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.

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.

[sujh]

@s3rius thanks
I updated to 1.0.6, cleared my redis and tested it again with the same steps mentioned above, the issue was still present. After shutdown and restarting the worker, the uncompleted tasks were not re-executed.

[sujh]

@s3rius
I tested it multiple times and met a very tircky issue. After restarting TaskIQ, it seems to take a very long time to fetch the previously unacknowledged tasks. There was one instance where I waited for almost seven minutes.

[spikeninja]

It's most likely because autoclaim part goes after xreadgroup in the 1.0.6 fix, so the unack messages are not processed at the beginning of the consuming

[s3rius]

Yes, we give a very big timeout for unacked messages in order to not suddenly claim long running tasks. In order to minimize the delay you can change idle_timeout argument and increase number of unacknowledged messages to be fetched on each xautoclaim call by updating unacknowledged_batch_size.