Add retry support for Celery email tasks - #687
Conversation
ℹ️ Recent review info⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (3)
📝 WalkthroughWalkthroughThe email-sending Celery task now retries transient failures up to three times, acknowledges work late, records the next execution time, and preserves retry propagation. Its failure test calls the task implementation directly, and ChangesEmail task retry handling
Environment ignore pattern
Estimated code review effort: 3 (Moderate) | ~20 minutes Possibly related issues
Suggested reviewers: 🚥 Pre-merge checks | ✅ 3 | ❌ 2❌ Failed checks (2 warnings)
✅ Passed checks (3 passed)
✨ Finishing Touches🧪 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: 3
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
backend/campaigns/tasks.py (1)
577-588: 🩺 Stability & Availability | 🔴 Critical | ⚡ Quick winAllow broker redeliveries to bypass the
next_execution_timenull check.While the newly added
acks_late=Trueparameter ensures that tasks are re-queued if a worker crashes, the atomic guard here will silently drop those redelivered tasks.If a worker crashes after line 583 but before successfully sending the email,
next_execution_timeremainsNone. When Celery subsequently redelivers the unacknowledged task, this exact query will fail to match becausenext_execution_timeis null, leaving the lead permanently stuck.You must bypass the
isnull=Falsecheck for broker redeliveries to ensure crashed tasks are actually resumed.💡 Proposed fix to respect `acks_late` redeliveries
- # Atomic guard: claim this send by nullifying next_execution_time. - # Only one concurrent caller can win; prevents duplicate sends. - claimed = CampaignLead.objects.filter( - id=campaign_lead_id, - current_step_id=step_id, - next_execution_time__isnull=False, - ).update(next_execution_time=None) + # Atomic guard: claim this send by nullifying next_execution_time. + # Allow Celery broker redeliveries (e.g. after a worker crash) to bypass the null check. + is_redelivered = self.request.delivery_info and self.request.delivery_info.get('redelivered') + + filter_kwargs = { + "id": campaign_lead_id, + "current_step_id": step_id, + } + if not is_redelivered: + filter_kwargs["next_execution_time__isnull"] = False + + claimed = CampaignLead.objects.filter(**filter_kwargs).update(next_execution_time=None) if not claimed: logger.info( f"Skipping duplicate send for {clead.lead.email} on step {step.step_order}" ) return🤖 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 `@backend/campaigns/tasks.py` around lines 577 - 588, Update the atomic claim logic around CampaignLead.objects.filter in the task to distinguish broker redeliveries from fresh deliveries and bypass the next_execution_time__isnull=False condition for redelivered tasks. Preserve the existing guard for fresh tasks so concurrent sends remain deduplicated, while allowing an acks_late redelivery to reclaim a lead whose next_execution_time is already null.
🤖 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 @.gitignore:
- Line 15: Replace the NUL-padded entry in .gitignore with a plain `.venv/`
pattern, preserving the intended virtualenv ignore rule.
In `@backend/campaigns/tasks.py`:
- Around line 626-637: Update the retry handling around self.retry and
MaxRetriesExceededError to use exponential countdowns of 1, 2, and 4 minutes
based on the current retry attempt instead of the fixed 15-minute
next_execution_time. When MaxRetriesExceededError occurs, place clead in the
established terminal failure status and clear or disable next_execution_time so
polling cannot enqueue it again. Record the permanent failure in EmailLog with
the relevant lead, error, and failure context.
In `@backend/campaigns/tests.py`:
- Around line 677-678: Update the test around send_email_step.run to wrap the
direct task invocation in assertRaises for celery.exceptions.Retry, while
preserving the existing send_gmail patch and subsequent database-state
assertions.
---
Outside diff comments:
In `@backend/campaigns/tasks.py`:
- Around line 577-588: Update the atomic claim logic around
CampaignLead.objects.filter in the task to distinguish broker redeliveries from
fresh deliveries and bypass the next_execution_time__isnull=False condition for
redelivered tasks. Preserve the existing guard for fresh tasks so concurrent
sends remain deduplicated, while allowing an acks_late redelivery to reclaim a
lead whose next_execution_time is already null.
🪄 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: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: 07d924d5-d2ac-4a45-899f-677438297ad0
📒 Files selected for processing (3)
.gitignorebackend/campaigns/tasks.pybackend/campaigns/tests.py
There was a problem hiding this comment.
Caution
Inline review comments failed to post. This is likely due to GitHub's internal server error or limits when posting large numbers of comments. If you are seeing this consistently it is likely a permissions issue. Please check "Moderation" -> "Code review limits" under your organization settings.
Actionable comments posted: 3
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
backend/campaigns/tasks.py (1)
577-588: 🩺 Stability & Availability | 🔴 Critical | ⚡ Quick winAllow broker redeliveries to bypass the
next_execution_timenull check.While the newly added
acks_late=Trueparameter ensures that tasks are re-queued if a worker crashes, the atomic guard here will silently drop those redelivered tasks.If a worker crashes after line 583 but before successfully sending the email,
next_execution_timeremainsNone. When Celery subsequently redelivers the unacknowledged task, this exact query will fail to match becausenext_execution_timeis null, leaving the lead permanently stuck.You must bypass the
isnull=Falsecheck for broker redeliveries to ensure crashed tasks are actually resumed.💡 Proposed fix to respect `acks_late` redeliveries
- # Atomic guard: claim this send by nullifying next_execution_time. - # Only one concurrent caller can win; prevents duplicate sends. - claimed = CampaignLead.objects.filter( - id=campaign_lead_id, - current_step_id=step_id, - next_execution_time__isnull=False, - ).update(next_execution_time=None) + # Atomic guard: claim this send by nullifying next_execution_time. + # Allow Celery broker redeliveries (e.g. after a worker crash) to bypass the null check. + is_redelivered = self.request.delivery_info and self.request.delivery_info.get('redelivered') + + filter_kwargs = { + "id": campaign_lead_id, + "current_step_id": step_id, + } + if not is_redelivered: + filter_kwargs["next_execution_time__isnull"] = False + + claimed = CampaignLead.objects.filter(**filter_kwargs).update(next_execution_time=None) if not claimed: logger.info( f"Skipping duplicate send for {clead.lead.email} on step {step.step_order}" ) return🤖 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 `@backend/campaigns/tasks.py` around lines 577 - 588, Update the atomic claim logic around CampaignLead.objects.filter in the task to distinguish broker redeliveries from fresh deliveries and bypass the next_execution_time__isnull=False condition for redelivered tasks. Preserve the existing guard for fresh tasks so concurrent sends remain deduplicated, while allowing an acks_late redelivery to reclaim a lead whose next_execution_time is already null.
🤖 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 @.gitignore:
- Line 15: Replace the NUL-padded entry in .gitignore with a plain `.venv/`
pattern, preserving the intended virtualenv ignore rule.
In `@backend/campaigns/tasks.py`:
- Around line 626-637: Update the retry handling around self.retry and
MaxRetriesExceededError to use exponential countdowns of 1, 2, and 4 minutes
based on the current retry attempt instead of the fixed 15-minute
next_execution_time. When MaxRetriesExceededError occurs, place clead in the
established terminal failure status and clear or disable next_execution_time so
polling cannot enqueue it again. Record the permanent failure in EmailLog with
the relevant lead, error, and failure context.
In `@backend/campaigns/tests.py`:
- Around line 677-678: Update the test around send_email_step.run to wrap the
direct task invocation in assertRaises for celery.exceptions.Retry, while
preserving the existing send_gmail patch and subsequent database-state
assertions.
---
Outside diff comments:
In `@backend/campaigns/tasks.py`:
- Around line 577-588: Update the atomic claim logic around
CampaignLead.objects.filter in the task to distinguish broker redeliveries from
fresh deliveries and bypass the next_execution_time__isnull=False condition for
redelivered tasks. Preserve the existing guard for fresh tasks so concurrent
sends remain deduplicated, while allowing an acks_late redelivery to reclaim a
lead whose next_execution_time is already null.
🪄 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: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: 07d924d5-d2ac-4a45-899f-677438297ad0
📒 Files selected for processing (3)
.gitignorebackend/campaigns/tasks.pybackend/campaigns/tests.py
🛑 Comments failed to post (3)
.gitignore (1)
15-15: 📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
python - <<'PY' from pathlib import Path line = Path(".gitignore").read_bytes().splitlines()[14] print(line) assert line == b".venv/", f"Unexpected bytes: {line!r}" PYRepository: Kuldeeep18/LeadOrbit
Length of output: 317
Replace the NUL-padded
.gitignoreentry with plain.venv/..gitignore:15contains embedded NUL bytes (.v\0e\0n\0v\0/), so the ignore rule won’t match the virtualenv directory.🤖 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 @.gitignore at line 15, Replace the NUL-padded entry in .gitignore with a plain `.venv/` pattern, preserving the intended virtualenv ignore rule.backend/campaigns/tasks.py (1)
626-637: 🩺 Stability & Availability | 🔴 Critical | ⚡ Quick win
Implement exponential backoff, prevent infinite retry loops, and record permanent failures.
The current error handling has two major flaws:
- Infinite Loop via Polling: When
MaxRetriesExceededErroris caught, the method returns while leavingnext_execution_timein the future (set on line 626). The background database polling loop will pick up this lead 15 minutes later and enqueue a new Celery task, bypassing themax_retrieslimit entirely and causing an infinite loop.- Missing PR Objectives: The code uses a static 15-minute delay instead of the requested "exponential retry delays of 1, 2, and 4 minutes", and it fails to record permanent failures in
EmailLogas specified in Issue#461.Ensure the lead is placed into a terminal status on max retries, calculate the correct countdown, and implement the
EmailLogrequirement.💡 Proposed fix
- clead.next_execution_time = timezone.now() + timedelta(minutes=15) - clead.save(update_fields=["next_execution_time"]) - - try: - self.retry(exc=send_err) - - except MaxRetriesExceededError: - logger.error( - f"Max retries exceeded for {clead.lead.email}" - ) + # Exponential backoff: 1, 2, 4 minutes + delay_minutes = 2 ** self.request.retries + + # Fallback guard: ensures the database polling loop won't race the Celery retry execution + clead.next_execution_time = timezone.now() + timedelta(minutes=delay_minutes + 5) + clead.save(update_fields=["next_execution_time"]) + + try: + self.retry(exc=send_err, countdown=delay_minutes * 60) + except MaxRetriesExceededError: + logger.error(f"Max retries exceeded for {clead.lead.email}") + # Permanently fail the lead to prevent the polling loop from blindly repeating it + clead.status = 'FAILED' # Or the appropriate error status + clead.next_execution_time = None + clead.save(update_fields=['status', 'next_execution_time']) + # TODO: Implement EmailLog creation for permanently failed sends as per `#461`🤖 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 `@backend/campaigns/tasks.py` around lines 626 - 637, Update the retry handling around self.retry and MaxRetriesExceededError to use exponential countdowns of 1, 2, and 4 minutes based on the current retry attempt instead of the fixed 15-minute next_execution_time. When MaxRetriesExceededError occurs, place clead in the established terminal failure status and clear or disable next_execution_time so polling cannot enqueue it again. Record the permanent failure in EmailLog with the relevant lead, error, and failure context.backend/campaigns/tests.py (1)
677-678: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
Catch the
Retryexception to prevent test crashes.Because
send_email_step.run()is called directly without a Celery worker,self.retry()will raise acelery.exceptions.Retryexception into the test scope. This unhandled exception will cause the test to crash before any assertions are executed.Wrap the execution in an
assertRaisesblock so the test can safely proceed to verify the database state.💡 Proposed fix
+ from celery.exceptions import Retry + with patch('campaigns.tasks.send_gmail', side_effect=Exception('gmail disabled')): - send_email_step.run(campaign_lead.id, email_step.id) + with self.assertRaises(Retry): + send_email_step.run(campaign_lead.id, email_step.id)📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.from celery.exceptions import Retry with patch('campaigns.tasks.send_gmail', side_effect=Exception('gmail disabled')): with self.assertRaises(Retry): send_email_step.run(campaign_lead.id, email_step.id)🤖 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 `@backend/campaigns/tests.py` around lines 677 - 678, Update the test around send_email_step.run to wrap the direct task invocation in assertRaises for celery.exceptions.Retry, while preserving the existing send_gmail patch and subsequent database-state assertions.
|
Hi @Kuldeeep18, could you please take a look at my code when you get a chance? I’ve added retry support for the Celery email tasks and updated the unit tests. Let me know if you have any feedback or suggestions! |
1 participant
Pull Request
🔗 Related Issue
Closes #461
📝 Summary of Changes
This PR improves the reliability of Celery email tasks by adding retry support.
Changes made:
bind=Trueto the Celery task.acks_late=True.max_retries=3.default_retry_delay=900(15 minutes).self.retry(...).next_execution_timescheduling behavior.🏷️ Type of Change
🧪 Testing
Tested the changes by running the campaign test suite.
Steps to test:
backenddirectory.python manage.py test campaigns.N/A (Backend change)
✅ Checklist
Summary by CodeRabbit