From 73466a11b8ad553e68def538e93cd277dbb89cba Mon Sep 17 00:00:00 2001 From: Michael Bunsen Date: Thu, 1 Oct 2026 16:44:00 -0700 Subject: [PATCH] fix(celery): keep task results in Redis and stop storing per-batch payloads Local, CI and the production example env files pinned CELERY_RESULT_BACKEND to rpc://, which keeps a long-lived RabbitMQ result consumer per process. RabbitMQ drops it after the 30 s heartbeat window when idle, so the next status read or wait() fails (job creation 500s; sync ML jobs end FAILURE after all saves succeeded). Removing the pin lets the settings derive Redis DB 1 from REDIS_URL. process_nats_pipeline_result now sets ignore_result=True: nothing reads its state, and with RESULT_EXTENDED each stored result copied the full ML payload. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01C7Xf6VPbwWtTumhjjF15g8 --- .envs/.ci/.django | 2 +- .envs/.local/.django | 2 +- .envs/.production/.django-example | 2 +- ami/jobs/tasks.py | 3 +++ ami/jobs/tests/test_tasks.py | 12 +++++++++++- config/settings/base.py | 5 ++--- 6 files changed, 19 insertions(+), 7 deletions(-) diff --git a/.envs/.ci/.django b/.envs/.ci/.django index 6577ebece..8d2a4cbab 100644 --- a/.envs/.ci/.django +++ b/.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 diff --git a/.envs/.local/.django b/.envs/.local/.django index 5f65ac07c..44e63a96a 100644 --- a/.envs/.local/.django +++ b/.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 diff --git a/.envs/.production/.django-example b/.envs/.production/.django-example index 1b8b6b0b8..03021bc7d 100644 --- a/.envs/.production/.django-example +++ b/.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 diff --git a/ami/jobs/tasks.py b/ami/jobs/tasks.py index b00213ebb..31910f112 100644 --- a/ami/jobs/tasks.py +++ b/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 diff --git a/ami/jobs/tests/test_tasks.py b/ami/jobs/tests/test_tasks.py index c2a6829a3..f755e44ef 100644 --- a/ami/jobs/tests/test_tasks.py +++ b/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. diff --git a/config/settings/base.py b/config/settings/base.py index 01883d0eb..74c435b1d 100644 --- a/config/settings/base.py +++ b/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