Allow user to configure a default job expiry-extra length (#303)

* Allow user to configure a default job expiry length

Previously, expires_extra_ms in arq/connections.py set a default expiry
of 1 day from when a job was expected to start to when it would expire
if no expiry was set in enqueue_job().

This commit keeps that as the default, but allows both ArqRedis and Worker
to have a larger default expiry configured, as self.expires_extra_ms.

expires_extra_ms is also moved to arq/constants.py along with other
shared constants.

* Add test for time-to-live of job keys based on expires_extra_ms

Author: isobelhooper

arq/connections.py

@@ -12,7 +12,7 @@
 from redis.asyncio.sentinel import Sentinel
 from redis.exceptions import RedisError, WatchError
 
-from .constants import default_queue_name, job_key_prefix, result_key_prefix
+from .constants import default_queue_name, expires_extra_ms, job_key_prefix, result_key_prefix
 from .jobs import Deserializer, Job, JobDef, JobResult, Serializer, deserialize_job, serialize_job
 from .utils import timestamp_ms, to_ms, to_unix_ms
 
@@ -71,10 +71,6 @@ def __repr__(self) -> str:
         return 'RedisSettings({})'.format(', '.join(f'{k}={v!r}' for k, v in self.__dict__.items()))
 
 
-# extra time after the job is expected to start when the job key should expire, 1 day in ms
-expires_extra_ms = 86_400_000
-
-
 if TYPE_CHECKING:
     BaseRedis = Redis[bytes]
 else:
@@ -89,6 +85,8 @@ class ArqRedis(BaseRedis):
     :param job_serializer: a function that serializes Python objects to bytes, defaults to pickle.dumps
     :param job_deserializer: a function that deserializes bytes into Python objects, defaults to pickle.loads
     :param default_queue_name: the default queue name to use, defaults to ``arq.queue``.
+    :param expires_extra_ms: the default length of time from when a job is expected to start
+     after which the job expires, defaults to 1 day in ms.
     :param kwargs: keyword arguments directly passed to ``redis.asyncio.Redis``.
     """
 
@@ -98,13 +96,15 @@ def __init__(
         job_serializer: Optional[Serializer] = None,
         job_deserializer: Optional[Deserializer] = None,
         default_queue_name: str = default_queue_name,
+        expires_extra_ms: int = expires_extra_ms,
         **kwargs: Any,
     ) -> None:
         self.job_serializer = job_serializer
         self.job_deserializer = job_deserializer
         self.default_queue_name = default_queue_name
         if pool_or_conn:
             kwargs['connection_pool'] = pool_or_conn
+        self.expires_extra_ms = expires_extra_ms
         super().__init__(**kwargs)
 
     async def enqueue_job(
@@ -157,7 +157,7 @@ async def enqueue_job(
             else:
                 score = enqueue_time_ms
 
-            expires_ms = expires_ms or score - enqueue_time_ms + expires_extra_ms
+            expires_ms = expires_ms or score - enqueue_time_ms + self.expires_extra_ms
 
             job = serialize_job(function, args, kwargs, _job_try, enqueue_time_ms, serializer=self.job_serializer)
             pipe.multi()
@@ -210,6 +210,7 @@ async def create_pool(
     job_serializer: Optional[Serializer] = None,
     job_deserializer: Optional[Deserializer] = None,
     default_queue_name: str = default_queue_name,
+    expires_extra_ms: int = expires_extra_ms,
 ) -> ArqRedis:
     """
     Create a new redis pool, retrying up to ``conn_retries`` times if the connection fails.
@@ -257,6 +258,7 @@ def pool_factory(*args: Any, **kwargs: Any) -> ArqRedis:
             pool.job_serializer = job_serializer
             pool.job_deserializer = job_deserializer
             pool.default_queue_name = default_queue_name
+            pool.expires_extra_ms = expires_extra_ms
             await pool.ping()
 
         except (ConnectionError, OSError, RedisError, asyncio.TimeoutError) as e:

arq/constants.py

@@ -13,3 +13,6 @@
 
 # used by `ms_to_datetime` to get the timezone
 timezone_env_vars = 'ARQ_TIMEZONE', 'arq_timezone', 'TIMEZONE', 'timezone'
+
+# extra time after the job is expected to start when the job key should expire, 1 day in ms
+expires_extra_ms = 86_400_000

arq/worker.py

@@ -19,6 +19,7 @@
     abort_job_max_age,
     abort_jobs_ss,
     default_queue_name,
+    expires_extra_ms,
     health_check_key_suffix,
     in_progress_key_prefix,
     job_key_prefix,
@@ -167,6 +168,8 @@ class Worker:
     :param max_burst_jobs: the maximum number of jobs to process in burst mode (disabled with negative values)
     :param job_serializer: a function that serializes Python objects to bytes, defaults to pickle.dumps
     :param job_deserializer: a function that deserializes bytes into Python objects, defaults to pickle.loads
+    :param expires_extra_ms: the default length of time from when a job is expected to start
+     after which the job expires, defaults to 1 day in ms.
     """
 
     def __init__(
@@ -198,6 +201,7 @@ def __init__(
         max_burst_jobs: int = -1,
         job_serializer: Optional[Serializer] = None,
         job_deserializer: Optional[Deserializer] = None,
+        expires_extra_ms: int = expires_extra_ms,
     ):
         self.functions: Dict[str, Union[Function, CronJob]] = {f.name: f for f in map(func, functions)}
         if queue_name is None:
@@ -261,6 +265,7 @@ def __init__(
         self.max_burst_jobs = max_burst_jobs
         self.job_serializer = job_serializer
         self.job_deserializer = job_deserializer
+        self.expires_extra_ms = expires_extra_ms
 
     def run(self) -> None:
         """
@@ -311,6 +316,7 @@ async def main(self) -> None:
                 job_deserializer=self.job_deserializer,
                 job_serializer=self.job_serializer,
                 default_queue_name=self.queue_name,
+                expires_extra_ms=self.expires_extra_ms,
             )
 
         logger.info('Starting worker for %d functions: %s', len(self.functions), ', '.join(self.functions))

tests/test_worker.py

@@ -10,7 +10,7 @@
 import pytest
 
 from arq.connections import ArqRedis, RedisSettings
-from arq.constants import abort_jobs_ss, default_queue_name, health_check_key_suffix, job_key_prefix
+from arq.constants import abort_jobs_ss, default_queue_name, expires_extra_ms, health_check_key_suffix, job_key_prefix
 from arq.jobs import Job, JobStatus
 from arq.worker import (
     FailedJobs,
@@ -308,6 +308,26 @@ async def test_job_expired_run_check(arq_redis: ArqRedis, worker, caplog):
     assert exc_info.value.job_results[0].result == JobExecutionFailed('job expired')
 
 
+@pytest.mark.parametrize(
+    'extra_job_expiry,wait_time',
+    [
+        (None, expires_extra_ms),
+        (1_000_000, 1_000_000),
+        (10_000_000, 10_000_000),
+        (999_999_999, 999_999_999),
+    ],
+)
+async def test_default_job_expiry(arq_redis: ArqRedis, worker, caplog, extra_job_expiry, wait_time):
+    """Test that jobs have a default expiry time based on the expires_extra_ms property in
+    ArqRedis."""
+    caplog.set_level(logging.INFO)
+    if extra_job_expiry is not None:
+        arq_redis.expires_extra_ms = extra_job_expiry
+    await arq_redis.enqueue_job('foobar', _job_id='testing')
+    time_to_live_ms = await arq_redis.pttl(job_key_prefix + 'testing')
+    assert time_to_live_ms == pytest.approx(wait_time)
+
+
 async def test_job_old(arq_redis: ArqRedis, worker, caplog):
     caplog.set_level(logging.INFO)
     await arq_redis.enqueue_job('foobar', _job_id='testing', _defer_by=-2)