Possibility of retrying after task timeout

[espdev]

We encountered some difficulties in managing the retrying logic.
Let's say we want to set a timeout for a task, but we also want retry to be triggered when a TimeoutError occurs.

As I understand it, we cannot control this behavior at present. The task is considered done when a TimeoutError occurs. According to the following code:

streaq/streaq/worker.py

Lines 1053 to 1058 in 8c221ba

	except TimeoutError as e:
	if not task.silent:
	logger.error(f"task {task.fn_name} … {task_id} timed out")
	result = e
	success = False
	done = True

But in many cases, it would be helpful to try running the task again after it ran out of time last time.

@Graeme22 how do you think about it?

[Graeme22]

You need to handle the TimeoutError in the task itself (or in a middleware) and raise a StreaqRetry instead.

[espdev]

Yes, that’s what I thought - the only way right now is to do it this way. Perhaps some mechanism at the worker logic level (customizable for each task) would be more convenient, middlewares are global for all tasks.

[Graeme22]

If you need to customize per task why not just handle the exceptions in the task?

[espdev]

TimeoutError exception raised by worker logic (when I set timeout for the task), I cannot handle it on the task level.

[Graeme22]

Ah I see, it can't be handled on the task level but the middleware is overkill because it's just the single task

[espdev]

Exactly. Right now, we only have one task for which we would like to have this behavior. In middleware, you need to filter by task name somehow. Furthermore, we control the number of attempts through a configuration parameter (in addition to max_tries), and we cannot control this in middleware; attempts will continue until max_tries is reached. In general, it's not very convenient.

[espdev]

The retrying logic works in a middleware because timeout check is also a middleware at the bottom level (fail_after) and we can handle it in the middleware. :)

streaq/streaq/worker.py

Lines 1009 to 1018 in 8c221ba

	async def _fn(*args: Any, **kwargs: Any) -> Any:
	with fail_after(to_seconds(task.timeout)):
	return await task.fn(*args, **kwargs)
	if not task.silent:
	logger.info(f"task {task.fn_name} □ {task_id} → worker {self.id}")
	wrapped = _fn
	for middleware in reversed(self.middlewares):
	wrapped = middleware(wrapped)

I can probably make a “middleware” for my task with the same timeout checking logic and handle it.

[Graeme22]

One thing you could do is instead of relying on the built-in timeout, add your own move_on_after block and then raise a StreaqRetry.

This is specific enough that I don't think it's worth the complexity of adding, but if you can come up with a simple fix I'm happy to look at a PR.

[espdev]

Yes, I will consider how this can be done in general terms. For now, I will create an additional wrapper specifically for that task.
Thank you!