Merge pull request #95 from walison17/refresh-escape

Add CACHEBACK_VALIDATE_JOB_REFRESH_NEEDED setting to skip already-fresh refreshes

Author: stephrdev

cacheback/base.py

@@ -314,6 +314,27 @@ def refresh(self, *args, **kwargs):
         self.store(self.key(*args, **kwargs), self.expiry(*args, **kwargs), result)
         return result
 
+    def should_refresh(self, *args, **kwargs):
+        """
+        Verify if the cache should be refreshed.
+
+        When ``CACHEBACK_VALIDATE_JOB_REFRESH_NEEDED`` is enabled, the refresh
+        is skipped if the cached value is still fresh (i.e. its ``lifetime``
+        has not yet been exceeded). Otherwise the refresh always runs.
+        """
+        if not getattr(settings, 'CACHEBACK_VALIDATE_JOB_REFRESH_NEEDED', False):
+            return True
+
+        expiry, data = self.cache.get(self.key(*args, **kwargs), (None, None))
+        if data is None:
+            return True
+
+        delta = expiry - time.time() - self.refresh_timeout
+        if delta > 0:
+            return False
+
+        return True
+
     def async_refresh(self, *args, **kwargs):
         """
         Trigger an asynchronous job to refresh the cache
@@ -481,10 +502,16 @@ def perform_async_refresh(cls, klass_str, obj_args, obj_kwargs, call_args, call_
         logger.info(
             "Using %s with constructor args %r and kwargs %r", klass_str, obj_args, obj_kwargs
         )
+
+        job = klass(*obj_args, **obj_kwargs)
+        if not job.should_refresh(*call_args, **call_kwargs):
+            logger.info('Refresh escaped, cache is already fresh.')
+            return
+
         logger.info("Calling refresh with args %r and kwargs %r", call_args, call_kwargs)
         start = time.time()
         try:
-            klass(*obj_args, **obj_kwargs).refresh(*call_args, **call_kwargs)
+            job.refresh(*call_args, **call_kwargs)
         except Exception as e:
             logger.exception("Error running job: '%s'", e)
         else:

docs/settings.rst

@@ -28,3 +28,21 @@ Make sure that the corresponding task queue is configured too.
 
 This specifies whether to ignore the result of the ``refresh_cache`` task
 and prevent Celery/RQ from storing it into its results backend.
+
+
+``CACHEBACK_VALIDATE_JOB_REFRESH_NEEDED``
+-----------------------------------------
+
+When enabled, the async refresh task verifies the cached value's expiry
+before running and skips the refresh if the cached value still has more
+than ``refresh_timeout`` seconds remaining before its expiry. This avoids
+redundant work when refresh tasks for the same key pile up in a slow
+worker queue: once one of them refreshes the cache, the remaining tasks
+short-circuit instead of re-fetching the same data.
+
+The ``refresh_timeout`` margin ensures that tasks are still allowed to
+run when the cache is about to expire, so a stale value isn't served
+because the next refresh was skipped too eagerly.
+
+Defaults to ``False`` to preserve the previous behavior of always
+refreshing when a task is picked up.

pyproject.toml

@@ -49,6 +49,7 @@ pytest-isort = ">=4.0"
 pytest-flake8 = ">=1.3"
 flake8 = ">=7"
 pytest-black =">=0.6"
+black = ">=25.0,<26"
 freezegun = ">=1.5"
 coverage = {version = ">=7.0", extras = ["toml"]}
 celery = ">=5.0"

tests/test_base_job.py

@@ -199,6 +199,44 @@ def test_should_stale_item_be_fetched_synchronously_reached(self):
     def test_should_stale_item_be_fetched_synchronously_not_reached(self):
         assert StaleDummyJob().should_stale_item_be_fetched_synchronously(300) is False
 
+    def test_should_refresh_setting_disabled(self, settings):
+        settings.CACHEBACK_VALIDATE_JOB_REFRESH_NEEDED = False
+        assert DummyJob().should_refresh('foo') is True
+
+    def test_should_refresh_cache_empty(self, settings):
+        settings.CACHEBACK_VALIDATE_JOB_REFRESH_NEEDED = True
+        assert DummyJob().should_refresh('foo') is True
+
+    def test_should_refresh_cache_fresh_with_margin(self, settings):
+        settings.CACHEBACK_VALIDATE_JOB_REFRESH_NEEDED = True
+        with freeze_time('2016-03-20 14:00'):
+            job = DummyJob()
+            job.refresh('foo')
+
+        # lifetime=600, refresh_timeout=60 -> skip while delta > 0
+        # at +539s remaining time is 61s, which is > refresh_timeout=60s
+        with freeze_time('2016-03-20 14:08:59'):
+            assert job.should_refresh('foo') is False
+
+    def test_should_refresh_cache_within_refresh_timeout(self, settings):
+        settings.CACHEBACK_VALIDATE_JOB_REFRESH_NEEDED = True
+        with freeze_time('2016-03-20 14:00'):
+            job = DummyJob()
+            job.refresh('foo')
+
+        # at +540s the remaining time equals refresh_timeout -> must refresh
+        with freeze_time('2016-03-20 14:09:00'):
+            assert job.should_refresh('foo') is True
+
+    def test_should_refresh_cache_expired(self, settings):
+        settings.CACHEBACK_VALIDATE_JOB_REFRESH_NEEDED = True
+        with freeze_time('2016-03-20 14:00'):
+            job = DummyJob()
+            job.refresh('foo')
+
+        with freeze_time('2016-03-20 14:11'):
+            assert job.should_refresh('foo') is True
+
     def test_key_no_args_no_kwargs(self):
         assert DummyJob().key() == 'tests.test_base_job.DummyJob'
 
@@ -291,3 +329,27 @@ def test_job_refresh_perform_error(self, logger_mock):
     def test_job_refresh(self):
         Job.perform_async_refresh('tests.test_base_job.EmptyDummyJob', (), {}, ('foo',), {})
         assert EmptyDummyJob().get('foo') is not None
+
+    @pytest.mark.redis_required
+    @mock.patch('tests.test_base_job.EmptyDummyJob.fetch', return_value='bar')
+    @pytest.mark.parametrize(
+        'validate_refresh_enabled, refresh_count',
+        [
+            (True, 1),
+            (False, 2),
+        ],
+    )
+    def test_validate_job_refresh_needed(
+        self, fetch_mock, rq_burst, settings, validate_refresh_enabled, refresh_count
+    ):
+        settings.CACHEBACK_VALIDATE_JOB_REFRESH_NEEDED = validate_refresh_enabled
+
+        job = EmptyDummyJob()
+        job.task_options = {'is_async': False}
+
+        job.async_refresh('foo')
+        job.async_refresh('foo')
+
+        rq_burst()
+
+        assert fetch_mock.call_count == refresh_count