Result wait timeouts follow system clock adjustments #704

Taskiq version

Source at f08a54ab3a8e4e16045c84379cf26eb4fa83d4b2.

Python version

3.12.13

What operating system are you using?

Linux

What happened?

AsyncTaskiqTask.wait_result and taskiq.gather measure elapsed timeout time with time.time(). A forward system clock adjustment can end the wait before its timeout has elapsed. A backward adjustment can delay the timeout.

I reproduced both directions with controlled clocks and a fake result backend. Nonpositive timeouts are unaffected.

Relevant log output

A forward jump from 100 to 10000 raises TaskiqResultTimeoutError immediately with a one-second timeout, although the next poll has the result available.

Broker initialization code

No broker or running worker is needed to reproduce the result-wait logic.

import asyncio
from unittest.mock import AsyncMock, patch

from taskiq.abc.result_backend import AsyncResultBackend
from taskiq.result import TaskiqResult
from taskiq.task import AsyncTaskiqTask

async def reproduce():
    backend = AsyncMock(spec=AsyncResultBackend)
    backend.is_result_ready.side_effect = [False, True]
    backend.get_result.return_value = TaskiqResult(
        is_err=False, return_value=42, execution_time=0
    )
    task = AsyncTaskiqTask("task", backend)
    with patch("taskiq.task.time", side_effect=[100, 10000]):
        await task.wait_result(timeout=1, check_interval=0)

asyncio.run(reproduce())

The timeout should use a monotonic clock so that wall-clock corrections do not change it.