Fix submission worker Celery queue for S3 retention tagging#5143
Conversation
Submission workers enqueue tag_submission_artifact_retention_tags after evaluation, but ECS task definitions never set CELERY_QUEUE_NAME. That left Celery task_default_queue as None and caused AttributeError in expand_destination when calling .delay(). - Pass CELERY_QUEUE_NAME through COMMON_SETTINGS_DICT and worker task defs - Stop forcing task_default_queue to None in evalai/celery.py - Fall back to synchronous S3 tagging when no Celery queue is configured - Add unit tests for enqueue and queue resolution helpers Co-authored-by: Rishabh Jain <rishabhjain2018@gmail.com>
WalkthroughThis PR adds ChangesCelery queue routing for retention tagging
Estimated code review effort: 3 (Moderate) | ~25 minutes 🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (3)
tests/unit/jobs/test_s3_retention.py (1)
181-195: 🎯 Functional Correctness | 🔵 Trivial | 💤 Low valueConsider testing the empty-string
CELERY_QUEUE_NAMEedge case.The queue resolution helper treats an empty-string env var as falsy and falls through to
app.conf.task_default_queue. A test withCELERY_QUEUE_NAME=""would verify this boundary and prevent regressions if the env-var check is ever changed to a presence-only check.✨ Optional test addition
+ `@patch.dict`("os.environ", {"CELERY_QUEUE_NAME": ""}, clear=True) + `@patch`("evalai.celery.app") + def test_get_celery_queue_for_retention_tagging_empty_env_falls_back( + self, mock_app + ): + mock_app.conf.task_default_queue = "celery_dev" + self.assertEqual( + "celery_dev", get_celery_queue_for_retention_tagging() + )🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/unit/jobs/test_s3_retention.py` around lines 181 - 195, Add a test for the empty-string CELERY_QUEUE_NAME edge case in get_celery_queue_for_retention_tagging: verify that when os.environ contains CELERY_QUEUE_NAME set to an empty string, the helper falls back to evalai.celery.app.conf.task_default_queue instead of treating it as a valid queue name. Reuse the existing test pattern alongside test_get_celery_queue_for_retention_tagging_prefers_env and test_get_celery_queue_for_retention_tagging_uses_app_default so the boundary behavior is covered.apps/jobs/s3_retention.py (1)
154-164: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winEnqueue logic correctly dispatches async or falls back to sync.
The
apply_asynccall with explicitqueue=celery_queueand thereturnguard preventing the sync fallback from running are both correct. The deferred import oftag_submission_artifact_retention_tagsavoids unnecessary imports when the sync path is taken.Consider wrapping
apply_asyncin a try/except to fall back to synchronous tagging if the broker is unavailable, improving resilience beyond the transition period:💡 Optional resilience improvement
celery_queue = get_celery_queue_for_retention_tagging() if celery_queue: from jobs.tasks import tag_submission_artifact_retention_tags - tag_submission_artifact_retention_tags.apply_async( - (submission.pk, list(artifact_paths)), - queue=celery_queue, - ) - return + try: + tag_submission_artifact_retention_tags.apply_async( + (submission.pk, list(artifact_paths)), + queue=celery_queue, + ) + return + except Exception: + logger.exception( + "Failed to enqueue retention tagging, " + "falling back to sync: submission_id=%s", + submission.pk, + ) tag_submission_artifacts_for_retention(submission, artifact_paths)🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@apps/jobs/s3_retention.py` around lines 154 - 164, The enqueue path in the retention tagging flow should gracefully fall back to synchronous processing if Celery dispatch fails. Update the logic around get_celery_queue_for_retention_tagging() and tag_submission_artifact_retention_tags.apply_async to catch broker/dispatch exceptions, log the failure, and then call tag_submission_artifacts_for_retention as the fallback. Keep the existing queue selection, deferred import, and early return behavior in the submission retention handler.evalai/celery.py (1)
9-14: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueFix correctly prevents
Nonetask_default_queue.Moving
config_from_objectbefore the queue selection and guarding thetask_default_queueassignment with a truthy check onCELERY_QUEUE_NAMEdirectly addresses the root cause of the production crash.Consider using a local variable to avoid the double
os.environlookup:♻️ Optional refactor for clarity
app.config_from_object("django.conf:settings") +celery_queue_name = os.environ.get("CELERY_QUEUE_NAME") if settings.DEBUG: app.conf.task_default_queue = "celery_dev" -elif os.environ.get("CELERY_QUEUE_NAME"): - app.conf.task_default_queue = os.environ["CELERY_QUEUE_NAME"] +elif celery_queue_name: + app.conf.task_default_queue = celery_queue_name🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@evalai/celery.py` around lines 9 - 14, The Celery queue selection in app/config_from_object and the task_default_queue assignment can still leave a None value in production. Update the settings logic around app.conf.task_default_queue so it only assigns CELERY_QUEUE_NAME when the environment value is truthy, and keep the DEBUG branch handling intact. Use a local variable for CELERY_QUEUE_NAME in this block to avoid repeated os.environ lookups and make the guard explicit.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@apps/challenges/aws_utils.py`:
- Line 164: The COMMON_SETTINGS_DICT entry for CELERY_QUEUE_NAME currently
passes through os.environ.get("CELERY_QUEUE_NAME") and can render unset values
as the literal string "None" in ECS templates. Update the settings-building
logic in aws_utils.py so refresh_worker_task_definitions() never receives an
unset queue name, either by omitting the CELERY_QUEUE_NAME key when it is
missing or by validating and failing fast before the task definitions are
refreshed. Keep the fix localized around COMMON_SETTINGS_DICT and the
refresh_worker_task_definitions() flow so evalai/celery.py and
apps/jobs/s3_retention.py do not consume an invalid queue name.
---
Nitpick comments:
In `@apps/jobs/s3_retention.py`:
- Around line 154-164: The enqueue path in the retention tagging flow should
gracefully fall back to synchronous processing if Celery dispatch fails. Update
the logic around get_celery_queue_for_retention_tagging() and
tag_submission_artifact_retention_tags.apply_async to catch broker/dispatch
exceptions, log the failure, and then call
tag_submission_artifacts_for_retention as the fallback. Keep the existing queue
selection, deferred import, and early return behavior in the submission
retention handler.
In `@evalai/celery.py`:
- Around line 9-14: The Celery queue selection in app/config_from_object and the
task_default_queue assignment can still leave a None value in production. Update
the settings logic around app.conf.task_default_queue so it only assigns
CELERY_QUEUE_NAME when the environment value is truthy, and keep the DEBUG
branch handling intact. Use a local variable for CELERY_QUEUE_NAME in this block
to avoid repeated os.environ lookups and make the guard explicit.
In `@tests/unit/jobs/test_s3_retention.py`:
- Around line 181-195: Add a test for the empty-string CELERY_QUEUE_NAME edge
case in get_celery_queue_for_retention_tagging: verify that when os.environ
contains CELERY_QUEUE_NAME set to an empty string, the helper falls back to
evalai.celery.app.conf.task_default_queue instead of treating it as a valid
queue name. Reuse the existing test pattern alongside
test_get_celery_queue_for_retention_tagging_prefers_env and
test_get_celery_queue_for_retention_tagging_uses_app_default so the boundary
behavior is covered.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro Plus
Run ID: f0f1c922-f0c7-41c9-af53-df8731eee5e6
📒 Files selected for processing (5)
apps/challenges/aws_utils.pyapps/challenges/task_definitions.pyapps/jobs/s3_retention.pyevalai/celery.pytests/unit/jobs/test_s3_retention.py
- Validate CELERY_QUEUE_NAME before building worker task definitions - Inject queue name at format time to avoid rendering None as "None" - Fall back to sync S3 tagging when Celery broker dispatch fails - Refactor celery.py queue selection with a local variable - Add tests for missing/empty queue name and broker dispatch fallback Co-authored-by: Rishabh Jain <rishabhjain2018@gmail.com>
|
Bugbot is not enabled for your account, so this pull request was not reviewed. Enable Bugbot in the Cursor dashboard to get automatic reviews on future PRs. |
- Format long @patch.dict decorators with black - Provide CELERY_QUEUE_NAME via autouse fixture for register task def tests Co-authored-by: Rishabh Jain <rishabhjain2018@gmail.com>
There was a problem hiding this comment.
🧹 Nitpick comments (1)
tests/unit/challenges/test_aws_utils.py (1)
6172-6246: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winVerify
CELERY_QUEUE_NAMEis passed through to the task definition template.The happy-path tests mock
task_definition.formatand assert the return value, but never inspect the format call's kwargs to confirmCELERY_QUEUE_NAMEis actually threaded into the template. If theupdated_settings["CELERY_QUEUE_NAME"] = celery_queue_nameline were accidentally removed, these tests would still pass silently.As per path instructions, tests should actually assert behavior and not silently pass. Consider adding an assertion on
mock_task_definition.format.call_argsto verify the queue name is present.♻️ Suggested assertion for each happy-path test
After each
build_task_definition_dictcall, add:task_def, error = build_task_definition_dict(challenge, "queue") self.assertIsNone(error) self.assertEqual(task_def, {"family": "queue"}) + format_kwargs = mock_task_definition.format.call_args.kwargs + self.assertEqual(format_kwargs.get("CELERY_QUEUE_NAME"), "evalai-celery") mock_task_definition.format.assert_called_once()🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/unit/challenges/test_aws_utils.py` around lines 6172 - 6246, The happy-path tests for build_task_definition_dict do not verify that CELERY_QUEUE_NAME is actually forwarded into the task definition template. Update the tests around mock_task_definition.format in both test_build_task_definition_dict_normal_challenge and test_build_task_definition_dict_code_upload_challenge to assert the format call received the queue name in its kwargs. This will ensure the updated_settings["CELERY_QUEUE_NAME"] assignment in build_task_definition_dict is covered and cannot be removed without failing tests.Source: Path instructions
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Nitpick comments:
In `@tests/unit/challenges/test_aws_utils.py`:
- Around line 6172-6246: The happy-path tests for build_task_definition_dict do
not verify that CELERY_QUEUE_NAME is actually forwarded into the task definition
template. Update the tests around mock_task_definition.format in both
test_build_task_definition_dict_normal_challenge and
test_build_task_definition_dict_code_upload_challenge to assert the format call
received the queue name in its kwargs. This will ensure the
updated_settings["CELERY_QUEUE_NAME"] assignment in build_task_definition_dict
is covered and cannot be removed without failing tests.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro Plus
Run ID: e92bd577-4a16-46ce-b1d0-2e7fca3d146a
📒 Files selected for processing (2)
tests/unit/challenges/test_aws_utils.pytests/unit/jobs/test_s3_retention.py
🚧 Files skipped from review as they are similar to previous changes (1)
- tests/unit/jobs/test_s3_retention.py
|
Bugbot is not enabled for your account, so this pull request was not reviewed. Enable Bugbot in the Cursor dashboard to get automatic reviews on future PRs. |
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## master #5143 +/- ##
==========================================
+ Coverage 91.20% 91.23% +0.03%
==========================================
Files 114 114
Lines 8865 8884 +19
==========================================
+ Hits 8085 8105 +20
+ Misses 780 779 -1
Flags with carried forward coverage won't be shown. Click here to find out more.
... and 1 file with indirect coverage changes
... and 1 file with indirect coverage changes Continue to review full report in Codecov by Harness.
🚀 New features to boost your workflow:
|
RishabhJain2018
deleted the
cursor/fix-celery-queue-submission-worker-44b5
branch
2 participants
Summary
Fixes the production
AttributeError: 'NoneType' object has no attribute 'pop'raised when submission workers callenqueue_submission_artifact_retention_tagging()after evaluation completes.The S3 retention feature enqueues
tag_submission_artifact_retention_tagsvia Celery, but ECS submission worker task definitions never setCELERY_QUEUE_NAME. That lefttask_default_queueasNone, causing Celery 4.3'sexpand_destination()to crash when routing the task.Changes
CELERY_QUEUE_NAMEinto submission worker containers (task_definitionandcontainer_definition_submission_worker), injected at build time with validation so unset values are never rendered as the literal string"None".evalai/celery.py— Configuretask_default_queueafterconfig_from_object()and only whenCELERY_QUEUE_NAMEis set (or in DEBUG).enqueue_submission_artifact_retention_tagging()— Useapply_async()with an explicit queue when available; fall back to synchronous tagging when no queue is configured or Celery dispatch fails.Deployment notes
refresh_worker_task_definitions(or redeploy workers) so running containers receiveCELERY_QUEUE_NAMEand resume async tagging via Celery.Testing
tests/unit/jobs/test_s3_retention.pyandtests/unit/challenges/test_aws_utils.py.Summary by CodeRabbit
New Features
CELERY_QUEUE_NAME, and requests fail with a clear error if it isn’t set.Bug Fixes
CELERY_QUEUE_NAMEis unset/empty.Tests
CELERY_QUEUE_NAMErequirements for ECS task definitions.