Stop jobs from failing when the stack has been idle, by keeping task results in Redis

.envs/.ci/.django

@@ -23,7 +23,7 @@ DEFAULT_PROCESSING_SERVICE_ENDPOINT=http://ml_backend:2000
 # RabbitMQ
 # ------------------------------------------------------------------------------
 CELERY_BROKER_URL=amqp://rabbituser:rabbitpass@rabbitmq:5672/
-CELERY_RESULT_BACKEND=rpc:// # Use RabbitMQ for results backend
+# CELERY_RESULT_BACKEND is unset on purpose: results go to Redis DB 1, derived from REDIS_URL.
 RABBITMQ_DEFAULT_USER=rabbituser
 RABBITMQ_DEFAULT_PASS=rabbitpass
 

.envs/.local/.django

@@ -30,7 +30,7 @@ CELERY_FLOWER_PASSWORD=BEQgmCtgyrFieKNoGTsux9YIye0I7P5Q7vEgfJD2C4jxmtHDetFaE2jhS
 
 # RabbitMQ
 CELERY_BROKER_URL=amqp://rabbituser:rabbitpass@rabbitmq:5672/
-CELERY_RESULT_BACKEND=rpc:// # Use RabbitMQ for results backend
+# CELERY_RESULT_BACKEND is unset on purpose: results go to Redis DB 1, derived from REDIS_URL.
 RABBITMQ_DEFAULT_USER=rabbituser
 RABBITMQ_DEFAULT_PASS=rabbitpass
 RABBITMQ_SERVER_ADDITIONAL_ERL_ARGS=-rabbit consumer_timeout 604800000

.envs/.production/.django-example

@@ -44,7 +44,7 @@ CELERY_FLOWER_PASSWORD=
 
 # RabbitMQ
 CELERY_BROKER_URL=
-CELERY_RESULT_BACKEND=rpc:// # Use RabbitMQ for results backend
+# CELERY_RESULT_BACKEND is unset on purpose: results go to Redis DB 1, derived from REDIS_URL.
 RABBITMQ_DEFAULT_USER=
 RABBITMQ_DEFAULT_PASS=
 RABBITMQ_SERVER_ADDITIONAL_ERL_ARGS=-rabbit consumer_timeout 604800000

ami/jobs/tasks.py

@@ -202,6 +202,9 @@ def _log_worker_availability(job) -> None:
 
 @celery_app.task(
     bind=True,
+    # Nothing reads this task's state, and RESULT_EXTENDED would store the full
+    # ML result payload in the result backend for every batch. See #1189.
+    ignore_result=True,
     # Retry on transient Redis/connection errors so a single connection reset
     # doesn't flip the job to FAILURE mid-processing. Backoff is capped at 15s
     # (half of NATS ack_wait = TASK_TTR = 30s, see nats_queue.py) so a retry

ami/jobs/tests/test_tasks.py

@@ -12,7 +12,7 @@
 from unittest.mock import AsyncMock, MagicMock, patch
 
 from django.core.cache import cache
-from django.test import TransactionTestCase
+from django.test import SimpleTestCase, TransactionTestCase
 from rest_framework.test import APITestCase
 
 from ami.base.serializers import reverse_with_params
@@ -515,6 +515,16 @@ def test_process_nats_pipeline_result_error_job_not_found(self, mock_manager_cla
         mock_manager.acknowledge_task.assert_called_once_with(reply_subject)
 
 
+class TestProcessNatsPipelineResultStoresNoResult(SimpleTestCase):
+    def test_task_ignores_its_result(self):
+        """The batch payload is saved to the database, not copied into the Celery result backend.
+
+        RESULT_EXTENDED stores task arguments with each result, which here is the full ML payload
+        for every batch. Nothing reads this task's state. See #1189.
+        """
+        self.assertTrue(process_nats_pipeline_result.ignore_result)
+
+
 class TestTaskFailureGuard(TransactionTestCase):
     """
     Bug C regression tests for the task_failure signal guard in update_job_failure.

config/settings/base.py

@@ -382,9 +382,8 @@ def _celery_result_backend_url(redis_url):
 # full ML result JSON as args. Measured on demo (298 keys, 2026-03-26):
 #   Median: 5 KB, Avg: 191 KB, Max: 2.1 MB per key
 #   Distribution: 29 <1KB, 195 1-10KB, 52 100KB-1MB, 22 >1MB
-# With thousands of tasks per job, this adds significant memory pressure.
-# TODO: consider disabling this or setting ignore_result=True on bulk tasks
-# like process_nats_pipeline_result to reduce result backend load. See #1189.
+# With thousands of tasks per job, this adds significant memory pressure, so
+# process_nats_pipeline_result sets ignore_result=True. See #1189.
 CELERY_RESULT_EXTENDED = True
 # https://docs.celeryq.dev/en/stable/userguide/configuration.html#result-backend-always-retry
 # https://github.com/celery/celery/pull/6122