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,9 +30,9 @@ 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".
+    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 (should be ignored by sentry).
 
     To bypass the cooldown for a specific invocation (e.g., operator-forced
     recovery), pass ``_cooldown_force=True`` as a kwarg through ``delay()``
@@ -73,6 +75,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()