feat(celery): cap worker-loss redeliveries to abandon repeatedly-killed tasks - #1199
feat(celery): cap worker-loss redeliveries to abandon repeatedly-killed tasks#1199ocervell wants to merge 3 commits into
Conversation
…ed tasks When a worker is killed mid-task (cgroup OOMKill of the child, or a node memory-pressure eviction), task_acks_late + task_reject_on_worker_lost cause the task to be redelivered with the same Celery id. A task that OOMs on every run would loop forever (OOM -> redeliver -> OOM) and the surrounding chord/workflow could never complete. Add task_max_retries (-1 = disabled). run_command counts redeliveries on the Celery result backend, keyed by the stable Celery id, and once the cap is exceeded abandons the task: it returns a forwarded result list with a FAILURE Error appended instead of re-running, so the chord proceeds and the workflow finishes with a clear 'abandoned after N retries' error (mirroring the inner memory-limit warning). The counter is backend-agnostic: it uses the result backend's generic key/value interface (atomic incr when available e.g. Redis/Memcached, else a get/set read-modify-write that every KV backend implements - filesystem, S3, GCS, ...). It is not tied to Redis or to any Secator data backend (Mongo/Postgres/SQLite). Backends with neither (database/ RPC) degrade safely to 'cap disabled'. Inert by default (task_max_retries=-1 and task_acks_late=False); enabling requires task_acks_late + task_reject_on_worker_lost and raising broker_visibility_timeout above the max task lifetime. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
|
Important Review skippedAuto incremental reviews are disabled on this repository. Please check the settings in the CodeRabbit UI or the ⚙️ Run configurationConfiguration used: Path: .coderabbit.yaml Review profile: CHILL Plan: Pro Run ID: You can disable this status message by setting the Use the checkbox below for a quick retry:
WalkthroughA new ChangesWorker-Loss Retry Cap
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~20 minutes Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ 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: 4
🧹 Nitpick comments (2)
tests/unit/test_celery.py (2)
344-365: ⚡ Quick winConsider logging exceptions in cleanup block.
The test cleanup properly uses a try-finally block, but the exception handler silently swallows all exceptions. While acceptable for cleanup code, logging the exception would aid debugging if cleanup fails.
📝 Proposed enhancement
finally: try: app.backend.delete(key) app.backend.delete(app.backend.get_key_for_task(f'worker-loss-{task_id}-other')) - except Exception: - pass + except Exception as e: + # Cleanup failures are non-fatal but worth noting for debugging + import logging + logging.debug(f'Test cleanup failed for {task_id}: {e}')🤖 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/test_celery.py` around lines 344 - 365, In the test_bump_worker_loss_count_get_set_fallback method, the cleanup code in the finally block contains an except Exception handler that silently passes without logging. Replace the bare pass statement with a logging call to capture exception details when the backend deletion fails, which will aid in debugging cleanup issues. Use an appropriate logger (such as logging.exception or a test logger) to record the exception that occurs during the app.backend.delete operations.Source: Linters/SAST tools
393-393: ⚡ Quick winUse
isinstanceinstead of accessing_typeattribute.Filtering errors by checking the private
_typeattribute is inconsistent with the established codebase pattern. Theerrors()method insecator/runners/_base.py:368-369usesisinstance(r, Error)for type checking.♻️ Proposed refactor for consistency
+ from secator.output_types import Error results = abandon_task('httpx', ['example.com'], {'context': {}}, []) - errors = [r for r in results if r._type == 'error'] + errors = [r for r in results if isinstance(r, Error)] self.assertEqual(len(errors), 1)🤖 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/test_celery.py` at line 393, Replace the private attribute access check in the list comprehension that filters errors to use isinstance() for consistency with the established codebase pattern. The line that checks r._type == 'error' should be refactored to use isinstance(r, Error) instead, matching the approach used elsewhere in secator/runners/_base.py. This improves code consistency and avoids relying on private attributes.
🤖 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 `@secator/celery.py`:
- Around line 307-310: The comparison logic in the task max retries check is off
by one. The bump_worker_loss_count function returns the total delivery count
(1-indexed), where 1 represents the initial delivery and subsequent values
represent redeliveries. To correctly compare against the retry cap, subtract 1
from the delivery_count to get the actual number of redeliveries before
comparing against CONFIG.celery.task_max_retries. Additionally, ensure that only
non-negative values of task_max_retries are treated as enabled caps so that a
value of 0 means zero retries allowed (not abandon on first delivery).
- Around line 234-237: The code stores a plain string value using
backend.set(key, str(count)), but Celery result backends expect serialized bytes
from an encoder, and byte-oriented KV backends may reject string payloads.
Encode the string counter value to bytes before passing it to backend.set by
using the appropriate encoder method (similar to how other values are encoded in
this context) to ensure compatibility with all backend types.
- Around line 221-237: The worker-loss counter keys created in
bump_worker_loss_count() do not have an expiry time set, causing them to
accumulate indefinitely on the backend and consume storage. After the
backend.incr(key) call in the atomic increment path and after the
backend.set(key, str(count)) call in the fallback get/set path, add expiry using
backend.expire(key, CONFIG.celery.result_expires) within try-except blocks to
gracefully handle backends that do not support the expire operation.
In `@tests/unit/test_celery.py`:
- Line 382: The continuation line containing patch.object(app.backend, 'get',
side_effect=NotImplementedError, create=True) is over-indented, which violates
the Flake8 E127 rule that is not in the project's ignore list. Reduce the
indentation of this continuation line to align it properly with the surrounding
context, ensuring it follows proper continuation line indentation standards for
the multi-line statement.
---
Nitpick comments:
In `@tests/unit/test_celery.py`:
- Around line 344-365: In the test_bump_worker_loss_count_get_set_fallback
method, the cleanup code in the finally block contains an except Exception
handler that silently passes without logging. Replace the bare pass statement
with a logging call to capture exception details when the backend deletion
fails, which will aid in debugging cleanup issues. Use an appropriate logger
(such as logging.exception or a test logger) to record the exception that occurs
during the app.backend.delete operations.
- Line 393: Replace the private attribute access check in the list comprehension
that filters errors to use isinstance() for consistency with the established
codebase pattern. The line that checks r._type == 'error' should be refactored
to use isinstance(r, Error) instead, matching the approach used elsewhere in
secator/runners/_base.py. This improves code consistency and avoids relying on
private attributes.
🪄 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
Run ID: 72edcaa0-2aa1-4d1c-b0f0-5f4784afe9e0
📒 Files selected for processing (3)
secator/celery.pysecator/config.pytests/unit/test_celery.py
| # Preferred: atomic incr (Redis, Memcached) — race-free. | ||
| try: | ||
| return backend.incr(key) | ||
| except NotImplementedError: | ||
| pass # Backend has no atomic counter; fall back to get/set below. | ||
| except Exception as e: | ||
| debug(f'worker-loss incr failed for {task_id}: {e}', sub='celery.state') | ||
| return 0 | ||
|
|
||
| # Fallback: get/set, implemented by every key/value result backend (filesystem, S3, GCS, ...). | ||
| # Worker-loss redeliveries of the same task id are sequential (only one attempt runs at a | ||
| # time), so a non-atomic read-modify-write is safe here. | ||
| try: | ||
| raw = backend.get(key) | ||
| count = (int(raw) if raw else 0) + 1 | ||
| backend.set(key, str(count)) | ||
| return count |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Description: Confirm whether counter keys are cleaned up in this PR and inspect Celery incr/expire support.
rg -n -C3 '\b(worker-loss|expire_worker_loss|delete\s*\()' --type=py
python - <<'PY'
import importlib
import inspect
for mod_name, cls_name in [
("celery.backends.redis", "RedisBackend"),
("celery.backends.cache", "CacheBackend"),
]:
try:
mod = importlib.import_module(mod_name)
cls = getattr(mod, cls_name)
print(f"\n## {mod_name}.{cls_name}.incr")
print(inspect.getsource(cls.incr))
if hasattr(cls, "expire"):
print(f"\n## {mod_name}.{cls_name}.expire")
print(inspect.getsource(cls.expire))
except Exception as exc:
print(f"\n## {mod_name}.{cls_name}: unavailable ({exc})")
PYRepository: freelabz/secator
Length of output: 4006
🏁 Script executed:
# Find all calls to bump_worker_loss_count to see cleanup patterns
rg -n 'bump_worker_loss_count' --type=py -A 10
# Search for cleanup/deletion after bump_worker_loss_count calls
rg -n 'delete.*worker-loss|cleanup' --type=py
# Check if there's any expiry logic in the codebase for these keys
rg -n 'expire|ttl' secator/celery.py --type=pyRepository: freelabz/secator
Length of output: 6112
🏁 Script executed:
# Read the full bump_worker_loss_count function and surrounding context
head -n 280 secator/celery.py | tail -n 100Repository: freelabz/secator
Length of output: 3460
🏁 Script executed:
# Search for where task completion/abandonment happens to see if cleanup occurs there
rg -n 'def.*task|@.*task' secator/celery.py -A 5 --type=py | head -100Repository: freelabz/secator
Length of output: 2869
🏁 Script executed:
# Look for cleanup on successful task completion or in task result cleanup
rg -n 'mark_completed|on_success|mark_finished' secator/celery.py -A 5 --type=py
# Check if result_expires applies to worker-loss keys anywhere
rg -n 'result_expires' secator/ --type=py
# Check backend implementations for TTL behavior
rg -n 'def delete|def expire' secator/celery.py --type=pyRepository: freelabz/secator
Length of output: 1297
🏁 Script executed:
# Check if there's any cleanup code added after bump_worker_loss_count call
rg -n -A 15 'bump_worker_loss_count' secator/celery.py --type=py | grep -A 15 'line 308'
# Check if expire/delete is called after bump_worker_loss_count anywhere
rg -n -A 5 'bump_worker_loss_count' secator/ --type=py | grep -E '(delete|expire|cleanup)'
# Verify if the function was modified to include expiry
cat secator/celery.py | sed -n '203,241p'Repository: freelabz/secator
Length of output: 1659
🏁 Script executed:
# Check the full context around the bump_worker_loss_count call at line 308
cat secator/celery.py | sed -n '300,320p'
# Check if there's any cleanup in the task completion path after the task runs
cat secator/celery.py | sed -n '278,350p'Repository: freelabz/secator
Length of output: 3399
🏁 Script executed:
# Check if the backend delete/expire methods are available and used elsewhere
rg -n 'backend\.(delete|expire)' secator/ --type=py
# Verify the full flow: after task completes, is there any cleanup?
rg -n 'def run_command' secator/celery.py -A 80 --type=py | tail -60Repository: freelabz/secator
Length of output: 2271
🏁 Script executed:
# Check if there's any automatic TTL/expiry applied to backend keys
rg -n 'get_key_for_task|backend.set' secator/celery.py -B 2 -A 2 --type=py
# Look for any cleanup scheduled or periodic tasks
rg -n '@.*periodic|cleanup|purge|expire' secator/ --type=py | head -20
# Check if the counter key is ever deleted in any path
rg -n 'delete.*worker-loss' secator/ --type=pyRepository: freelabz/secator
Length of output: 2061
🏁 Script executed:
# Check the Celery backend API documentation by looking at how set/incr are used
python3 -c "
import inspect
try:
from celery.backends.base import BaseBackend
print('BaseBackend.set signature:')
print(inspect.signature(BaseBackend.set))
print('\nBaseBackend.incr signature:')
print(inspect.signature(BaseBackend.incr))
except ImportError:
print('Celery not available in sandbox, checking method signatures in codebase')
"
# Check if backend.set is called with TTL anywhere in the codebase
rg -n 'backend\.set\(' secator/ -A 1 --type=pyRepository: freelabz/secator
Length of output: 222
Set TTL on worker-loss counter keys to prevent unbounded accumulation.
The bump_worker_loss_count() function increments/sets a custom worker-loss-{task_id} counter key (lines 221–237) but never expires it. With task capping enabled, each task creates a counter key that persists indefinitely on the backend, even after the task completes successfully or is abandoned. Over time, this accumulates stale keys, consuming backend storage.
Add an expiry aligned with CONFIG.celery.result_expires after incr and set calls, with graceful fallback for backends without expire support:
Suggested fix
+def expire_worker_loss_key(backend, key):
+ """Expire a worker-loss counter key if the backend supports it."""
+ expire = getattr(backend, 'expire', None)
+ if callable(expire) and CONFIG.celery.result_expires > 0:
+ try:
+ expire(key, CONFIG.celery.result_expires)
+ except NotImplementedError:
+ pass
+ except Exception as e:
+ debug(f'worker-loss expire failed: {e}', sub='celery.state')
+
# Preferred: atomic incr (Redis, Memcached) — race-free.
try:
- return backend.incr(key)
+ count = backend.incr(key)
+ expire_worker_loss_key(backend, key)
+ return count
except NotImplementedError:
pass # Backend has no atomic counter; fall back to get/set below.
except Exception as e:
debug(f'worker-loss incr failed for {task_id}: {e}', sub='celery.state')
return 0
# Fallback: get/set, implemented by every key/value result backend (filesystem, S3, GCS, ...).
try:
raw = backend.get(key)
count = (int(raw) if raw else 0) + 1
backend.set(key, str(count))
+ expire_worker_loss_key(backend, key)
return count🧰 Tools
🪛 Ruff (0.15.17)
[warning] 226-226: Do not catch blind exception: Exception
(BLE001)
🤖 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 `@secator/celery.py` around lines 221 - 237, The worker-loss counter keys
created in bump_worker_loss_count() do not have an expiry time set, causing them
to accumulate indefinitely on the backend and consume storage. After the
backend.incr(key) call in the atomic increment path and after the
backend.set(key, str(count)) call in the fallback get/set path, add expiry using
backend.expire(key, CONFIG.celery.result_expires) within try-except blocks to
gracefully handle backends that do not support the expire operation.
| raw = backend.get(key) | ||
| count = (int(raw) if raw else 0) + 1 | ||
| backend.set(key, str(count)) | ||
| return count |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Description: Inspect the installed Celery backend set() implementations for payload type expectations.
python - <<'PY'
import importlib
import inspect
import importlib.metadata
try:
print("celery", importlib.metadata.version("celery"))
except Exception as exc:
print("Could not determine Celery version:", exc)
for mod_name, cls_name in [
("celery.backends.filesystem", "FilesystemBackend"),
("celery.backends.s3", "S3Backend"),
("celery.backends.gcs", "GCSBackend"),
]:
try:
mod = importlib.import_module(mod_name)
cls = getattr(mod, cls_name)
print(f"\n## {mod_name}.{cls_name}.set")
print(inspect.getsource(cls.set))
except Exception as exc:
print(f"\n## {mod_name}.{cls_name}: unavailable ({exc})")
PYRepository: freelabz/secator
Length of output: 374
🏁 Script executed:
#!/bin/bash
# Find and read the secator/celery.py file, focusing on the bump_worker_loss_count function
fd -t f celery.py | head -5Repository: freelabz/secator
Length of output: 163
🏁 Script executed:
#!/bin/bash
# Read the full context around lines 203-240 in secator/celery.py
if [ -f "secator/celery.py" ]; then
wc -l secator/celery.py
echo "---"
sed -n '200,245p' secator/celery.py
fiRepository: freelabz/secator
Length of output: 1814
🌐 Web query:
Celery result backend set method API str bytes payload type
💡 Result:
In Celery's result backend architecture, the set method is a mandatory abstract method defined in base backend classes (such as KeyValueStoreBackend) that handles the actual persistence of task data [1][2][3]. When Celery stores a result, it typically goes through a process of encoding the data into a serialized payload before passing it to the backend. The internal _store_result method calls self.encode(meta), which produces a bytes object representing the serialized task metadata [1][4]. This encoded bytes payload is then passed to _set_with_state, which delegates to the backend's set(key, value) method [2][4]. While the abstract interface for set(self, key, value) does not strictly enforce the type of the value at the Python interface level, it is designed to receive the serialized output of the encoder [2][5]. In practice, because Celery backends often use serializers (like JSON or pickle) that return bytes, the payload handled by set is expected to be bytes in most production implementations [1][5]. Key takeaways: - The set method is responsible for persisting data, typically as bytes [1][2]. - Data is serialized using self.encode() before reaching the set method [1][4]. - Backends that do not handle bytes natively (e.g., some that might expect strings) may perform their own conversions, but the core Celery contract revolves around the serialized bytes representation produced by the configured serializer [1][3][5].
Citations:
- 1: https://docs.celeryq.dev/en/stable/%5Fmodules/celery/backends/base.html
- 2: https://github.com/celery/celery/blob/main/celery/backends/base.py
- 3: https://docs.celeryq.dev/en/stable/internals/reference/celery.backends.base.html
- 4: https://docs.celeryq.dev/en/main/_modules/celery/backends/base.html
- 5: https://github.com/sbdchd/celery-types/blob/main/celery-stubs/backends/base.pyi
Encode the fallback counter value before backend.set.
Line 236 stores a str; Celery result backends are designed to handle serialized bytes from the encoder, and byte-oriented KV backends (filesystem, S3, GCS) can reject string payloads. On failure, the broad exception handler silently returns 0, disabling the cap. Store an encoded byte value instead.
🐛 Proposed fix
- backend.set(key, str(count))
+ backend.set(key, str(count).encode())📝 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.
| raw = backend.get(key) | |
| count = (int(raw) if raw else 0) + 1 | |
| backend.set(key, str(count)) | |
| return count | |
| raw = backend.get(key) | |
| count = (int(raw) if raw else 0) + 1 | |
| backend.set(key, str(count).encode()) | |
| return count |
🤖 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 `@secator/celery.py` around lines 234 - 237, The code stores a plain string
value using backend.set(key, str(count)), but Celery result backends expect
serialized bytes from an encoder, and byte-oriented KV backends may reject
string payloads. Encode the string counter value to bytes before passing it to
backend.set by using the appropriate encoder method (similar to how other values
are encoded in this context) to ensure compatibility with all backend types.
| if CONFIG.celery.task_max_retries != -1 and CONFIG.celery.task_acks_late: | ||
| delivery_count = bump_worker_loss_count(self.request.id) | ||
| if delivery_count > CONFIG.celery.task_max_retries: | ||
| return abandon_task(name, targets, opts, results) |
There was a problem hiding this comment.
Compare against redeliveries, not total deliveries.
bump_worker_loss_count() returns 1 for the initial delivery, so task_max_retries = 0 abandons before the task ever runs. Treat only non-negative caps as enabled and compare delivery_count - 1 to the retry cap.
🐛 Proposed fix
- if CONFIG.celery.task_max_retries != -1 and CONFIG.celery.task_acks_late:
+ if CONFIG.celery.task_max_retries >= 0 and CONFIG.celery.task_acks_late:
delivery_count = bump_worker_loss_count(self.request.id)
- if delivery_count > CONFIG.celery.task_max_retries:
+ redelivery_count = max(delivery_count - 1, 0)
+ if redelivery_count > CONFIG.celery.task_max_retries:
return abandon_task(name, targets, opts, results)🤖 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 `@secator/celery.py` around lines 307 - 310, The comparison logic in the task
max retries check is off by one. The bump_worker_loss_count function returns the
total delivery count (1-indexed), where 1 represents the initial delivery and
subsequent values represent redeliveries. To correctly compare against the retry
cap, subtract 1 from the delivery_count to get the actual number of redeliveries
before comparing against CONFIG.celery.task_max_retries. Additionally, ensure
that only non-negative values of task_max_retries are treated as enabled caps so
that a value of 0 means zero retries allowed (not abandon on first delivery).
| from secator.celery import app, bump_worker_loss_count | ||
|
|
||
| with patch.object(app.backend, 'incr', side_effect=NotImplementedError, create=True), \ | ||
| patch.object(app.backend, 'get', side_effect=NotImplementedError, create=True): |
There was a problem hiding this comment.
Fix continuation line indentation.
The continuation line is over-indented. Flake8 E127 is not in the project's ignore list per coding guidelines.
🔧 Proposed fix
with patch.object(app.backend, 'incr', side_effect=NotImplementedError, create=True), \
- patch.object(app.backend, 'get', side_effect=NotImplementedError, create=True):
+ patch.object(app.backend, 'get', side_effect=NotImplementedError, create=True):
self.assertEqual(bump_worker_loss_count('task-abc'), 0)As per coding guidelines, the project uses flake8 linting with specific ignored rules (W191, E101, E128, E265, W605), but E127 is not ignored.
📝 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.
| patch.object(app.backend, 'get', side_effect=NotImplementedError, create=True): | |
| with patch.object(app.backend, 'incr', side_effect=NotImplementedError, create=True), \ | |
| patch.object(app.backend, 'get', side_effect=NotImplementedError, create=True): | |
| self.assertEqual(bump_worker_loss_count('task-abc'), 0) |
🧰 Tools
🪛 Flake8 (7.3.0)
[error] 382-382: continuation line over-indented for visual indent
(E127)
🤖 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/test_celery.py` at line 382, The continuation line containing
patch.object(app.backend, 'get', side_effect=NotImplementedError, create=True)
is over-indented, which violates the Flake8 E127 rule that is not in the
project's ignore list. Reduce the indentation of this continuation line to align
it properly with the surrounding context, ensuring it follows proper
continuation line indentation standards for the multi-line statement.
Sources: Coding guidelines, Linters/SAST tools
…loss cancel flag Completes the worker-loss redelivery cap with the review fixes + the L1 delivery flag: - Off-by-one: the initial delivery no longer counts as a retry. Extracted worker_loss_retries_exhausted(delivery_count, max_retries) — redeliveries = delivery_count - 1, so task_max_retries=3 allows the initial run + 3 redeliveries before abandoning (was 2). Unit-tested at the boundary. - Key expiry: worker-loss counter keys now get a best-effort TTL tied to result_expires (via _expire_worker_loss_key) on both the atomic-incr and the get/set paths, so they don't accumulate forever. - L1: add worker_cancel_long_running_tasks_on_connection_loss config flag (default False) wired into app.conf, alongside the existing task_acks_late / task_reject_on_worker_lost. Enables clean redelivery of a connection-lost worker's in-flight tasks. Inert until enabled in deployment. Note: the terminal-doc no-op guard (a redelivered task whose runner doc is already done) is deferred — it depends on stable runner identity (on_build, the L2 PR) to find 'its' doc, and is ineffective without it. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
|
Pushed the review fixes + the L1 delivery flag (commit
Deferred: the terminal-doc no-op guard (a redelivered task whose runner doc is already |
The abandon message said 'after N retries' using task_max_retries, which read
oddly at max_retries=0 ('after 0 retries') because a redelivery still occurs
even with 0 retries — the broker redelivers a lost task under acks_late; that
is NOT a re-run of the work. Thread the actual delivery_count through and
report it separately from the cap:
Task httpx abandoned after 5 delivery attempts (retry cap: 3; worker
repeatedly lost — likely OOM kill or node eviction).
Reads sensibly for any cap value (incl. 0).
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
ocervell
added
the
feature:worker-reliability
|
Superseded — this work is already merged into |
…ed tasks (#1284) **Re-extracted from the closed #1199** — this generic Celery worker-reliability fix was folded into the `ai-resiliency` branch (#1241, AI-task hardening, base `canary`) and its standalone PR closed. It's general worker-pool infra (nothing AI-specific), so it belongs on `main` on its own timeline. Cherry-picked clean onto current `main`. ## What it does Under `task_acks_late` + `task_reject_on_worker_lost` (prod config), a task that dies with its worker (SIGKILL / OOM / node eviction) is redelivered with the **same Celery id**. A task that OOMs on *every* run would loop forever (OOM → redeliver → OOM), blocking its chord indefinitely. This caps redeliveries: `run_command` counts worker-loss redeliveries (Redis `celery-task-meta-worker-loss-<id>`); once `task_max_retries` is exceeded it **abandons** the task with a clear `abandoned after N delivery attempts` result so the chord/workflow can finish. `task_max_retries=-1` disables the cap (default). Also adds the connection-loss cancel flag. ## Validation Exercised end-to-end earlier (prefork child-loss → requeue → abandon at cap, 6/6). **`tests/unit/test_celery.py`: 23 passed** on current `main`; flake8 clean. ## Note `feature:worker-reliability`. Companion PR extracts the eviction-finalize half (was #1203). Prod workers run `-P prefork -c 1`, so the fast child-loss path is live. 🤖 Generated with [Claude Code](https://claude.com/claude-code) <!-- This is an auto-generated comment: release notes by coderabbit.ai --> ## Summary by CodeRabbit * **New Features** * Added a limit for repeated worker-loss retries so long-running jobs can stop retrying after a configurable cap. * Introduced an option to cancel in-flight tasks when the broker connection is lost, helping prevent stuck background work. * **Bug Fixes** * Improved handling of interrupted tasks so workflows can complete with a clear failure result instead of hanging indefinitely. * Added safer retry counting to better track redelivered tasks across worker losses. <!-- end of auto-generated comment: release notes by coderabbit.ai --> --------- Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
1 participant
Problem
When a worker is killed mid-task — a cgroup OOMKill of the child process, or a node memory-pressure eviction — the task never returns a result. With
task_acks_late+task_reject_on_worker_lost, the broker redelivers the task (same Celery id) to a fresh worker. That's the desired recovery, except for a task that OOMs on every run (e.g. a memory spike the innertask_memory_limit_mbmonitor doesn't catch in time): it loops forever — OOM → redeliver → OOM — and the surrounding chord/workflow can never complete.This was hit in prod: a headless
katanatask onworker-largewas evicted (The node was low on resource: memory), the Job failed (BackoffLimitExceeded), and the workflow stalled.Change
Add
celery.task_max_retries(-1= disabled).run_commandcounts redeliveries (keyed by the stable Celery id) and, once the cap is exceeded, abandons the task instead of re-running it: it returns a forwarded result list with aFAILUREErrorappended ("Task <name> abandoned after N retries (worker repeatedly lost — likely OOM kill or node eviction)."), so the chord proceeds and the workflow finishes cleanly — mirroring the existing inner memory-limit warning.Backend-agnostic counter
The counter lives on the Celery result backend's generic key/value interface, not Redis internals and not any Secator data backend (Mongo/Postgres/SQLite):
incrwhen the backend provides it (Redis, Memcached) — race-free;get/setread-modify-write otherwise (filesystem, S3, GCS, cache, …) — safe because same-task redeliveries are sequential;Safety / rollout
Inert by default —
task_max_retries=-1andtask_acks_late=Falseship unchanged, so this commit changes no behavior. Enabling the cap requires, at deploy time:task_acks_late=1+task_reject_on_worker_lost=1,task_max_retriesset (e.g.3),broker_visibility_timeoutraised above the max task lifetime (tasks run up toactiveDeadlineSeconds=10800); leaving it at the default 1h would duplicate-execute long tasks onceacks_lateis on.Tests
tests/unit/test_celery.py— get/set fallback (real filesystem backend), atomic-incr path, unsupported-backend degradation, and the abandon path returning aFAILUREerror. Full file: 22 passing.Scope note
This fully covers the child-OOMKill retry loop. A whole-pod eviction kills the parent too, so its redelivery comes via the slower Redis visibility-timeout restore; that case is primarily addressed by task-profile sizing (see #1198) + Guaranteed QoS. Externally un-sticking an already-stuck chord is a separate follow-up (Layer 3).
🤖 Generated with Claude Code
Summary by CodeRabbit