Raise ResultNotFound when Job.result() finds no job and no result (#364)

* Raise ResultNotFound when Job.result() finds no job and no result

Job.result() assumed that a result will eventually show up in Redis.
This assumption does not hold for jobs which are not in the queue
and jobs which do not keep result. In such cases Job.result() would
hang forever (or until the optional timeout is reached).

This PR changes Job.result() so that it raises a new exception
ResultNotFound, rather than waiting forever, if result is not found
and the job is not in the queue.

* Subclass ResultNotFound from RuntimeError

Co-authored-by: Samuel Colvin <[email protected]>

* Add message when raising ResultNotFound

* Improve docstring for Job.result()

* Handle ResultNotFound in Job.abort()

* Update arq/jobs.py

Co-authored-by: Samuel Colvin <[email protected]>

Author: iamlikeme

arq/jobs.py

@@ -18,6 +18,10 @@
 Deserializer = Callable[[bytes], Dict[str, Any]]
 
 
+class ResultNotFound(RuntimeError):
+    pass
+
+
 class JobStatus(str, Enum):
     """
     Enum of job statuses.
@@ -82,8 +86,10 @@ async def result(
         self, timeout: Optional[float] = None, *, poll_delay: float = 0.5, pole_delay: float = None
     ) -> Any:
         """
-        Get the result of the job, including waiting if it's not yet available. If the job raised an exception,
-        it will be raised here.
+        Get the result of the job or, if the job raised an exception, reraise it.
+
+        This function waits for the result if it's not yet available and the job is
+        present in the queue. Otherwise ``ResultNotFound`` is raised.
 
         :param timeout: maximum time to wait for the job result before raising ``TimeoutError``, will wait forever
         :param poll_delay: how often to poll redis for the job result
@@ -96,15 +102,24 @@ async def result(
             poll_delay = pole_delay
 
         async for delay in poll(poll_delay):
-            info = await self.result_info()
-            if info:
-                result = info.result
+            async with self._redis.pipeline(transaction=True):
+                v = await self._redis.get(result_key_prefix + self.job_id)
+                s = await self._redis.zscore(self._queue_name, self.job_id)
+
+            if v:
+                info = deserialize_result(v, deserializer=self._deserializer)
                 if info.success:
-                    return result
-                elif isinstance(result, (Exception, asyncio.CancelledError)):
-                    raise result
+                    return info.result
+                elif isinstance(info.result, (Exception, asyncio.CancelledError)):
+                    raise info.result
                 else:
-                    raise SerializationError(result)
+                    raise SerializationError(info.result)
+            elif s is None:
+                raise ResultNotFound(
+                    'Not waiting for job result because the job is not in queue. '
+                    'Is the worker function configured to keep result?'
+                )
+
             if timeout is not None and delay > timeout:
                 raise asyncio.TimeoutError()
 
@@ -174,6 +189,9 @@ async def abort(self, *, timeout: Optional[float] = None, poll_delay: float = 0.
             await self.result(timeout=timeout, poll_delay=poll_delay)
         except asyncio.CancelledError:
             return True
+        except ResultNotFound:
+            # We do not know if the job was cancelled or not
+            return False
         else:
             return False
 

tests/test_jobs.py

@@ -7,7 +7,15 @@
 from arq import Worker, func
 from arq.connections import ArqRedis, RedisSettings, create_pool
 from arq.constants import default_queue_name, in_progress_key_prefix, job_key_prefix, result_key_prefix
-from arq.jobs import DeserializationError, Job, JobResult, JobStatus, deserialize_job_raw, serialize_result
+from arq.jobs import (
+    DeserializationError,
+    Job,
+    JobResult,
+    JobStatus,
+    ResultNotFound,
+    deserialize_job_raw,
+    serialize_result,
+)
 
 
 async def test_job_in_progress(arq_redis: ArqRedis):
@@ -25,11 +33,36 @@ async def test_unknown(arq_redis: ArqRedis):
 
 
 async def test_result_timeout(arq_redis: ArqRedis):
-    j = Job('foobar', arq_redis)
+    j = await arq_redis.enqueue_job('foobar')
     with pytest.raises(asyncio.TimeoutError):
         await j.result(0.1, poll_delay=0)
 
 
+async def test_result_not_found(arq_redis: ArqRedis):
+    j = Job('foobar', arq_redis)
+    with pytest.raises(ResultNotFound):
+        await j.result()
+
+
+async def test_result_when_job_does_not_keep_result(arq_redis: ArqRedis, worker):
+    async def foobar(ctx):
+        pass
+
+    worker: Worker = worker(functions=[func(foobar, name='foobar', keep_result=0)])
+    j = await arq_redis.enqueue_job('foobar')
+
+    result_call = asyncio.Task(j.result())
+
+    _, pending = await asyncio.wait([result_call], timeout=0.1)
+    assert pending == {result_call}
+
+    await worker.main()
+
+    with pytest.raises(ResultNotFound):
+        # Job has completed and did not store any result
+        await asyncio.wait_for(result_call, timeout=5)
+
+
 async def test_enqueue_job(arq_redis: ArqRedis, worker, queue_name=default_queue_name):
     async def foobar(ctx, *args, **kwargs):
         return 42

tests/test_worker.py

@@ -810,6 +810,11 @@ async def wait_and_abort(job, delay=0.1):
     assert worker.job_tasks == {}
 
 
+async def test_abort_job_which_is_not_in_queue(arq_redis: ArqRedis):
+    job = Job(job_id='testing', redis=arq_redis)
+    assert await job.abort() is False
+
+
 async def test_abort_job_before(arq_redis: ArqRedis, worker, caplog, loop):
     async def longfunc(ctx):
         await asyncio.sleep(3600)