Repository: python-arq/arq (language: python)

Issue:
No exception on job timeout
In v0.20 when job is timed out `asyncio.TimeoutError` is not raised. Instead there is log message: `job_name cancelled, will be run again` and job is restarted immediately.
Is it intentional? Having some exception on timeout and allow app to decide how to handle it seems more appropriate to me.  
No exception on job timeout
In v0.20 when job is timed out `asyncio.TimeoutError` is not raised. Instead there is log message: `job_name cancelled, will be run again` and job is restarted immediately.
Is it intentional? Having some exception on timeout and allow app to decide how to handle it seems more appropriate to me.  


Discussion / hints:
I think `asyncio.TimeoutError` is raised, it's just caught and the job is re-run.

If you want to do something different, you might be able to catch it inside the job?
Inside the job there is `asyncio.CancelledError` on timeout now. 
I guess it's because in https://github.com/samuelcolvin/arq/pull/212/
`async with async_timeout.timeout(timeout_s):` 
was replaced with 
`cancel_handler = self.loop.call_at(self.loop.time() + timeout_s, task.cancel)`
Humm, the the logic I implemented was copied from async-timeout, so I would have expected the behaviour to be the same.

Guess I'm wrong...
https://github.com/aio-libs/async-timeout/blob/master/async_timeout/__init__.py#L191-L194
`async-timeout` checks exception type in `__exit__`
and raises `asyncio.TimeoutError` if it's `asyncio.CancelledError` and task was cancelled by `async-timeout`.
I can't find similar logic in `arq`
Oh, missed that. Happy to accept a PR if we can fix it.
Is it ok to just return `async-timeout` back or there is some deep reason behind removing it?
yes, we need access to the task to add it to `self.job_tasks`
For stuff like this, it's worth reading the code rather than asking questions, it's relatively clear in the code:

https://github.com/samuelcolvin/arq/blob/8ee04f96934684c9b8f8278df0530ae9fad1be4a/arq/worker.py#L520-L535
Actually inside a job it's always been `asyncio.CancelledError` on timeout. The difference only in retries. In 0.20 job is retried and previously exception was just logged.
Humm, I think we should revert this to just log the error, you can always catch the `CancelledError` and raise a retry exception.

Would you create a PR?
I think `asyncio.TimeoutError` is raised, it's just caught and the job is re-run.

If you want to do something different, you might be able to catch it inside the job?
Inside the job there is `asyncio.CancelledError` on timeout now. 
I guess it's because in https://github.com/samuelcolvin/arq/pull/212/
`async with async_timeout.timeout(timeout_s):` 
was replaced with 
`cancel_handler = self.loop.call_at(self.loop.time() + timeout_s, task.cancel)`
Humm, the the logic I implemented was copied from async-timeout, so I would have expected the behaviour to be the same.

Guess I'm wrong...
https://github.com/aio-libs/async-timeout/blob/master/async_timeout/__init__.py#L191-L194
`async-timeout` checks exception type in `__exit__`
and raises `asyncio.TimeoutError` if it's `asyncio.CancelledError` and task was cancelled by `async-timeout`.
I can't find similar logic in `arq`
Oh, missed that. Happy to accept a PR if we can fix it.
Is it ok to just return `async-timeout` back or there is some deep reason behind removing it?
yes, we need access to the task to add it to `self.job_tasks`
For stuff like this, it's worth reading the code rather than asking questions, it's relatively clear in the code:

https://github.com/samuelcolvin/arq/blob/8ee04f96934684c9b8f8278df0530ae9fad1be4a/arq/worker.py#L520-L535
Actually inside a job it's always been `asyncio.CancelledError` on timeout. The difference only in retries. In 0.20 job is retried and previously exception was just logged.
Humm, I think we should revert this to just log the error, you can always catch the `CancelledError` and raise a retry exception.

Would you create a PR?

Fix patch:
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 2a0930ec..e24b2e7e 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -79,7 +79,7 @@ jobs:
 
     - run: coverage xml
 
-    - uses: codecov/codecov-action@v1.4.1
+    - uses: codecov/codecov-action@v1.5.2
       with:
         file: ./coverage.xml
         env_vars: PYTHON,OS
diff --git a/arq/worker.py b/arq/worker.py
index b29c73bd..680105c3 100644
--- a/arq/worker.py
+++ b/arq/worker.py
@@ -519,10 +519,9 @@ async def job_failed(exc: BaseException) -> None:
             logger.info('%6.2fs → %s(%s)%s', (start_ms - enqueue_time_ms) / 1000, ref, s, extra)
             self.job_tasks[job_id] = task = self.loop.create_task(function.coroutine(ctx, *args, **kwargs))
 
-            cancel_handler = self.loop.call_at(self.loop.time() + timeout_s, task.cancel)
             # run repr(result) and extra inside try/except as they can raise exceptions
             try:
-                result = await task
+                result = await asyncio.wait_for(task, timeout_s)
             except (Exception, asyncio.CancelledError) as e:
                 exc_extra = getattr(e, 'extra', None)
                 if callable(exc_extra):
@@ -532,7 +531,6 @@ async def job_failed(exc: BaseException) -> None:
                 result_str = '' if result is None else truncate(repr(result))
             finally:
                 del self.job_tasks[job_id]
-                cancel_handler.cancel()
 
         except (Exception, asyncio.CancelledError) as e:
             finished_ms = timestamp_ms()


Test patch:
diff --git a/tests/test_worker.py b/tests/test_worker.py
index f4ea3bdc..e74716fb 100644
--- a/tests/test_worker.py
+++ b/tests/test_worker.py
@@ -817,3 +817,21 @@ async def wait_and_abort(job, delay=0.1):
     assert worker.aborting_tasks == set()
     assert worker.tasks == {}
     assert worker.job_tasks == {}
+
+
+async def test_job_timeout(arq_redis: ArqRedis, worker, caplog):
+    async def longfunc(ctx):
+        await asyncio.sleep(0.3)
+
+    caplog.set_level(logging.ERROR)
+    await arq_redis.enqueue_job('longfunc', _job_id='testing')
+    worker: Worker = worker(functions=[func(longfunc, name='longfunc')], job_timeout=0.2, poll_delay=0.1)
+    assert worker.jobs_complete == 0
+    assert worker.jobs_failed == 0
+    assert worker.jobs_retried == 0
+    await worker.main()
+    assert worker.jobs_complete == 0
+    assert worker.jobs_failed == 1
+    assert worker.jobs_retried == 0
+    log = re.sub(r'\d+.\d\ds', 'X.XXs', '\n'.join(r.message for r in caplog.records))
+    assert 'X.XXs ! testing:longfunc failed, TimeoutError:' in log