Raise Reject if cooldown in effect (feedback) and force ignore cooldown in mgmt commands by default

Author: mbertrand

learning_resources/management/commands/backpopulate_mit_edx_data.py

@@ -37,12 +37,6 @@ def add_arguments(self, parser):
             help="If provided, use this file as the source of program API data",
             default=None,
         )
-        parser.add_argument(
-            "--force",
-            dest="force",
-            action="store_true",
-            help="Bypass the task cooldown and run immediately",
-        )
         super().add_arguments(parser)
 
     def handle(self, *args, **options):  # noqa: ARG002
@@ -59,7 +53,7 @@ def handle(self, *args, **options):  # noqa: ARG002
             task = get_mit_edx_data.delay(
                 options["api_course_datafile"],
                 options["api_program_datafile"],
-                _cooldown_force=options.get("force", False),
+                _cooldown_force=True,
             )
             self.stdout.write(f"Started task {task} to get MIT edX course data")
             self.stdout.write("Waiting on task...")
@@ -70,4 +64,6 @@ def handle(self, *args, **options):  # noqa: ARG002
             self.stdout.write(
                 f"Population of MIT edX data finished, took {total_seconds} seconds"
             )
-            self.write_population_result(count)
+            self.stdout.write(
+                f"Populated {count} resources. See celery logs for details."
+            )

learning_resources/management/commands/backpopulate_mitxonline_data.py

@@ -25,12 +25,6 @@ def add_arguments(self, parser):
             action="store_true",
             help="Delete all existing records first",
         )
-        parser.add_argument(
-            "--force",
-            dest="force",
-            action="store_true",
-            help="Bypass the task cooldown and run immediately",
-        )
         super().add_arguments(parser)
 
     def handle(self, *args, **options):  # noqa: ARG002
@@ -44,9 +38,7 @@ def handle(self, *args, **options):  # noqa: ARG002
             ):
                 resource_delete_actions(learning_resource)
         else:
-            task = get_mitxonline_data.delay(
-                _cooldown_force=options.get("force", False),
-            )
+            task = get_mitxonline_data.delay(_cooldown_force=True)
             self.stdout.write(f"Started task {task} to get MITx Online course data")
             self.stdout.write("Waiting on task...")
             start = now_in_utc()
@@ -56,4 +48,6 @@ def handle(self, *args, **options):  # noqa: ARG002
             self.stdout.write(
                 f"Population of MITX Online data finished, took {total_seconds} seconds"
             )
-            self.write_population_result(count)
+            self.stdout.write(
+                f"Populated {count} resources. See celery logs for details."
+            )

learning_resources/management/commands/backpopulate_oll_data.py

@@ -31,12 +31,6 @@ def add_arguments(self, parser):
             help="If provided, use this google sheets id to get the data",
             default=None,
         )
-        parser.add_argument(
-            "--force",
-            dest="force",
-            action="store_true",
-            help="Bypass the task cooldown and run immediately",
-        )
         super().add_arguments(parser)
 
     def handle(self, *args, **options):  # noqa: ARG002
@@ -53,7 +47,7 @@ def handle(self, *args, **options):  # noqa: ARG002
         else:
             task = get_oll_data.delay(
                 sheets_id=options["sheets_id"],
-                _cooldown_force=options.get("force", False),
+                _cooldown_force=True,
             )
             self.stdout.write(f"Started task {task} to get oll course data")
             self.stdout.write("Waiting on task...")
@@ -64,4 +58,6 @@ def handle(self, *args, **options):  # noqa: ARG002
             self.stdout.write(
                 f"Population of oll data finished, took {total_seconds} seconds"
             )
-            self.write_population_result(count)
+            self.stdout.write(
+                f"Populated {count} resources. See celery logs for details."
+            )

learning_resources/management/commands/backpopulate_xpro_data.py

@@ -25,12 +25,6 @@ def add_arguments(self, parser):
             action="store_true",
             help="Delete all existing records first",
         )
-        parser.add_argument(
-            "--force",
-            dest="force",
-            action="store_true",
-            help="Bypass the task cooldown and run immediately",
-        )
         super().add_arguments(parser)
 
     def handle(self, *args, **options):  # noqa: ARG002
@@ -45,9 +39,7 @@ def handle(self, *args, **options):  # noqa: ARG002
             ):
                 resource_delete_actions(learning_resource)
         else:
-            task = get_xpro_data.delay(
-                _cooldown_force=options.get("force", False),
-            )
+            task = get_xpro_data.delay(_cooldown_force=True)
             self.stdout.write(f"Started task {task} to get xpro course data")
             self.stdout.write("Waiting on task...")
             start = now_in_utc()
@@ -57,4 +49,6 @@ def handle(self, *args, **options):  # noqa: ARG002
             self.stdout.write(
                 f"Population of xpro data finished, took {total_seconds} seconds"
             )
-            self.write_population_result(count)
+            self.stdout.write(
+                f"Populated {count} resources. See celery logs for details."
+            )

learning_resources/management/commands/mixins.py

@@ -23,17 +23,6 @@ def configure_test_resources(self, options):
                 test_mode=True, published=False
             )
 
-    def write_population_result(self, count):
-        """Write the final result line for a backpopulate command."""
-        if count is None:
-            self.stdout.write(
-                "Task skipped (cooldown active). See celery logs for details."
-            )
-        else:
-            self.stdout.write(
-                f"Populated {count} resources. See celery logs for details."
-            )
-
 
 class ConfirmDeleteMixin:
     """

main/decorators.py

@@ -5,6 +5,8 @@
 from collections.abc import Callable
 from functools import wraps
 
+from celery import current_task, states
+from celery.exceptions import Reject
 from django.core.cache import caches
 
 log = logging.getLogger(__name__)
@@ -28,16 +30,23 @@ def cooldown_task(
     Place this *below* ``@app.task`` so the cooldown runs on the worker, not
     on the enqueuing process.
 
-    Dropped calls return ``None``. If the wrapped function may itself return
-    ``None``, callers cannot disambiguate — annotate as ``int | None`` (or
-    similar) and treat ``None`` as "skipped".
-
-    To bypass the cooldown for a specific invocation (e.g., operator-forced
-    recovery), pass ``_cooldown_force=True`` as a kwarg through ``delay()``
-    or ``apply_async``. The wrapper consumes it before calling the wrapped
-    function and refreshes the lock so subsequent calls are still gated.
-    This is race-free relative to clearing the lock from outside, which has
-    a window between clear and enqueue where another worker can reacquire.
+    When a call is skipped from inside a Celery worker, the task's state is
+    explicitly set to ``REJECTED`` in the result backend and
+    ``Reject(requeue=False)`` is raised. Celery's own reject handling does
+    not write a state to the backend, so the ``update_state`` is what makes
+    ``AsyncResult.status`` observable as ``REJECTED`` rather than
+    ``PENDING``. ``Reject`` is a Celery control-flow exception and is not
+    reported to Sentry. Direct (non-task) calls return ``None``.
+
+    The cooldown is intended to gate scheduled (Celery Beat) invocations.
+    Operator-initiated calls (management commands, ad-hoc ``delay()`` from a
+    shell) should pass ``_cooldown_force=True`` as a kwarg through ``delay()``
+    or ``apply_async`` so they are not skipped — and so synchronous callers
+    using ``task.get()`` do not block waiting on a skipped run. The wrapper
+    consumes the kwarg before calling the wrapped function and refreshes the
+    lock so subsequent scheduled calls are still gated. This is race-free
+    relative to clearing the lock from outside, which has a window between
+    clear and enqueue where another worker can reacquire.
 
     The wrapper also exposes ``clear_cooldown(*args, **kwargs)`` which
     deletes the lock key. Useful for operational debugging from a shell;
@@ -73,6 +82,13 @@ def wrapper(*args, **kwargs):
                 caches["redis"].set(lock_key, "1", timeout=wait_time)
             elif not caches["redis"].add(lock_key, "1", timeout=wait_time):
                 log.info("Skipping %s: cooldown active (%ss)", lock_key, wait_time)
+                if current_task and not current_task.request.called_directly:
+                    current_task.update_state(
+                        state=states.REJECTED,
+                        meta={"reason": "cooldown", "key": lock_key},
+                    )
+                    reason = "cooldown active"
+                    raise Reject(reason, requeue=False)
                 return None
             return func(*args, **kwargs)
 

main/decorators_test.py

@@ -1,6 +1,7 @@
 """Tests for main.decorators"""
 
 import pytest
+from celery.exceptions import Reject
 
 from main.decorators import cooldown_task
 
@@ -144,3 +145,39 @@ def my_task(**kwargs):
     my_task(_cooldown_force=True)
     assert my_task() is None
     mock_redis.add.assert_called_once()
+
+
+def test_cooldown_task_raises_reject_inside_celery_worker(mock_redis, mocker):
+    """
+    When a skip happens inside a real Celery task run, write REJECTED to the
+    result backend and raise Reject so the run is observable as REJECTED
+    rather than PENDING/SUCCESS.
+    """
+    mock_redis.add.return_value = False
+    mock_task = mocker.patch("main.decorators.current_task")
+    mock_task.request.called_directly = False
+
+    @cooldown_task(wait_time=60)
+    def my_task():
+        return 1
+
+    with pytest.raises(Reject) as exc_info:
+        my_task()
+    assert exc_info.value.requeue is False
+    mock_task.update_state.assert_called_once()
+    assert mock_task.update_state.call_args.kwargs["state"] == "REJECTED"
+    assert mock_task.update_state.call_args.kwargs["meta"]["reason"] == "cooldown"
+
+
+def test_cooldown_task_returns_none_when_called_directly(mock_redis, mocker):
+    """Direct (non-worker) skipped calls return None — no Reject, no state write."""
+    mock_redis.add.return_value = False
+    mock_task = mocker.patch("main.decorators.current_task")
+    mock_task.request.called_directly = True
+
+    @cooldown_task(wait_time=60)
+    def my_task():
+        return 1
+
+    assert my_task() is None
+    mock_task.update_state.assert_not_called()