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

Issue:
Cron job termination error
Hi, 
when worker is terminated with CTRL+C while any cron job is running there is exception: 
`aioredis.errors.MultiExecError: ('MultiExecError errors:', [TypeError("delete() missing 1 required positional argument: 'key'")])`

It happens here: https://github.com/samuelcolvin/arq/blob/8ee04f96934684c9b8f8278df0530ae9fad1be4a/arq/worker.py#L625 because `delete_keys` is empty. Obvious fix would be to check if `delete_keys` is not empty before doing `tr.delete(*delete_keys) ` but I'm not sure if it's expected situation or problem is deeper.


Discussion / hints:
I have the same problem but not only when I want to terminate the process. In my case it's critical to have `timeout` for a `cron` job. And the code below:
```python
import asyncio
from arq import cron

async def run_regularly(ctx):
    await asyncio.sleep(90)

class WorkerSettings:
    cron_jobs = [cron(run_regularly, timeout=55)]
```
produce the next result:
```sh
09:51:25: Starting worker for 1 functions: cron:run_regularly
09:51:25: redis_version=6.2.5 mem_usage=915.47K clients_connected=4 db_keys=0
09:52:00:   0.39s → cron:run_regularly()
09:52:55:  55.06s ↻ cron:run_regularly cancelled, will be run again
Traceback (most recent call last):
  File "/Users/mmelnichuk/Repository/pythonProject1/venv/bin/arq", line 8, in <module>
    sys.exit(cli())
  File "/Users/mmelnichuk/Repository/pythonProject1/venv/lib/python3.9/site-packages/click/core.py", line 1137, in __call__
    return self.main(*args, **kwargs)
  File "/Users/mmelnichuk/Repository/pythonProject1/venv/lib/python3.9/site-packages/click/core.py", line 1062, in main
    rv = self.invoke(ctx)
  File "/Users/mmelnichuk/Repository/pythonProject1/venv/lib/python3.9/site-packages/click/core.py", line 1404, in invoke
    return ctx.invoke(self.callback, **ctx.params)
  File "/Users/mmelnichuk/Repository/pythonProject1/venv/lib/python3.9/site-packages/click/core.py", line 763, in invoke
    return __callback(*args, **kwargs)
  File "/Users/mmelnichuk/Repository/pythonProject1/venv/lib/python3.9/site-packages/arq/cli.py", line 48, in cli
    run_worker(worker_settings_, **kwargs)
  File "/Users/mmelnichuk/Repository/pythonProject1/venv/lib/python3.9/site-packages/arq/worker.py", line 758, in run_worker
    worker.run()
  File "/Users/mmelnichuk/Repository/pythonProject1/venv/lib/python3.9/site-packages/arq/worker.py", line 256, in run
    self.loop.run_until_complete(self.main_task)
  File "/usr/local/Cellar/python@3.9/3.9.5/Frameworks/Python.framework/Versions/3.9/lib/python3.9/asyncio/base_events.py", line 642, in run_until_complete
    return future.result()
  File "/Users/mmelnichuk/Repository/pythonProject1/venv/lib/python3.9/site-packages/arq/worker.py", line 308, in main
    await self._poll_iteration()
  File "/Users/mmelnichuk/Repository/pythonProject1/venv/lib/python3.9/site-packages/arq/worker.py", line 346, in _poll_iteration
    t.result()
  File "/Users/mmelnichuk/Repository/pythonProject1/venv/lib/python3.9/site-packages/arq/worker.py", line 590, in run_job
    await asyncio.shield(
  File "/Users/mmelnichuk/Repository/pythonProject1/venv/lib/python3.9/site-packages/arq/worker.py", line 628, in finish_job
    await tr.execute()
  File "/Users/mmelnichuk/Repository/pythonProject1/venv/lib/python3.9/site-packages/aioredis/commands/transaction.py", line 176, in execute
    return await self._do_execute(
  File "/Users/mmelnichuk/Repository/pythonProject1/venv/lib/python3.9/site-packages/aioredis/commands/transaction.py", line 296, in _do_execute
    return (await self._gather_result(return_exceptions))
  File "/Users/mmelnichuk/Repository/pythonProject1/venv/lib/python3.9/site-packages/aioredis/commands/transaction.py", line 198, in _gather_result
    raise self.error_class(errors)
aioredis.errors.MultiExecError: ('MultiExecError errors:', [TypeError("delete() missing 1 required positional argument: 'key'")])
```
I've found the way how to bypass this bug/situation. If I set `retry_jobs = False` for a worker then the code above will work without error. But I want to have possibility to retry the jobs :)

Fix patch:
diff --git a/arq/worker.py b/arq/worker.py
index 680105c3..7b3973fa 100644
--- a/arq/worker.py
+++ b/arq/worker.py
@@ -620,7 +620,8 @@ async def finish_job(
                 tr.zrem(self.queue_name, job_id)
             elif incr_score:
                 tr.zincrby(self.queue_name, incr_score, job_id)
-            tr.delete(*delete_keys)
+            if delete_keys:
+                tr.delete(*delete_keys)
             await tr.execute()
 
     async def finish_failed_job(self, job_id: str, result_data: Optional[bytes]) -> None:


Test patch:
diff --git a/tests/test_cron.py b/tests/test_cron.py
index 6f603e21..8df3da15 100644
--- a/tests/test_cron.py
+++ b/tests/test_cron.py
@@ -1,3 +1,4 @@
+import asyncio
 import logging
 import re
 from datetime import datetime, timedelta
@@ -5,6 +6,7 @@
 
 import pytest
 
+import arq
 from arq import Worker
 from arq.constants import in_progress_key_prefix
 from arq.cron import cron, next_cron
@@ -142,3 +144,19 @@ async def test_repr():
 async def test_str_function():
     cj = cron('asyncio.sleep', hour=1, run_at_startup=True)
     assert str(cj).startswith('<CronJob name=cron:asyncio.sleep coroutine=<function sleep at')
+
+
+async def test_cron_cancelled(worker, mocker):
+    mocker.patch.object(arq.worker, 'keep_cronjob_progress', 0.1)
+
+    async def try_sleep(ctx):
+        if ctx['job_try'] == 1:
+            raise asyncio.CancelledError
+
+    worker: Worker = worker(
+        cron_jobs=[cron(try_sleep, microsecond=20, run_at_startup=True, max_tries=2)], poll_delay=0.01,
+    )
+    await worker.main()
+    assert worker.jobs_complete == 1
+    assert worker.jobs_retried == 1
+    assert worker.jobs_failed == 0