feat(celery): cap worker-loss redeliveries to abandon repeatedly-killed tasks

[ocervell]

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

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.

[coderabbitai]

Important

Review skipped

Auto incremental reviews are disabled on this repository.

Please check the settings in the CodeRabbit UI or the .coderabbit.yaml file in this repository. To trigger a single review, invoke the @coderabbitai review command.

⚙️ Run configuration

Configuration used: Path: .coderabbit.yaml

Review profile: CHILL

Plan: Pro

Run ID: b5391454-bdca-40a0-a5c4-295cad783ff8

You can disable this status message by setting the reviews.review_status to false in the CodeRabbit configuration file.

Use the checkbox below for a quick retry:

• 🔍 Trigger review

Walkthrough

Adds a configurable Celery worker-loss retry cap. New task_max_retries and worker_cancel_long_running_tasks_on_connection_loss config fields are introduced, along with helper functions to count worker-loss redeliveries per task and abandon tasks (returning an Error result) once the cap is exceeded, wired into run_command. Unit tests cover the new behavior.

Changes

Worker-loss retry cap

Layer / File(s)	Summary
Config fields for retry cap secator/config.py	Adds task_max_retries (default -1, unlimited) and worker_cancel_long_running_tasks_on_connection_loss (default False) Celery config fields with explanatory comments.
Worker-loss counter and abandonment helpers secator/celery.py	Imports Error output type, wires the new connection-loss config into the Celery app, and adds helpers to expire counter keys, bump per-task worker-loss counts (atomic incr or get/set fallback), check retry exhaustion, and abandon tasks with a failure Error result.
run_command retry-cap enforcement secator/celery.py	run_command bumps the worker-loss count for the current task and returns abandon_task(...) immediately when the redelivery cap is exceeded and task_acks_late is enabled.
Unit tests for retry cap tests/unit/test_celery.py	Adds TestWorkerLossRetryCap covering counter increment fallback and atomic paths, unsupported-backend degradation, abandon_task error output, exhaustion boundary logic, and config wiring.

Estimated code review effort: 3 (Moderate) | ~25 minutes

Sequence Diagram(s)

sequenceDiagram
  participant Broker
  participant Worker
  participant run_command
  participant Backend

  Broker->>Worker: redeliver task (worker lost)
  Worker->>run_command: execute task
  run_command->>Backend: bump_worker_loss_count(task_id)
  Backend-->>run_command: delivery count
  run_command->>run_command: worker_loss_retries_exhausted(count, max_retries)
  alt exhausted
    run_command->>run_command: abandon_task(...)
    run_command-->>Broker: return Error result (abandoned)
  else not exhausted
    run_command-->>Broker: continue normal execution
  end

Loading

Poem

A worker fell, then fell again,
OOM'd and lost, an endless chain—
This bunny counts each sad redeliver,
Then caps the loop, no more shall shiver.
Hop, abandon, chord completes at last! 🐇✨

🚥 Pre-merge checks | ✅ 4 | ❌ 1

❌ Failed checks (1 warning)

Check name	Status	Explanation	Resolution
Out of Scope Changes check	⚠️ Warning	The new connection-loss cancel flag is extra to the retry-cap issue and is not part of the linked issue's requested behavior.	Remove or split out the worker_cancel_long_running_tasks_on_connection_loss config change if it is not intended to be part of this issue.

✅ Passed checks (4 passed)

Check name	Status	Explanation
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Title check	✅ Passed	The title is concise and accurately summarizes the main Celery retry-cap change.
Linked Issues check	✅ Passed	The changes implement the requested worker-loss retry cap, backend-agnostic counter, and abandonment path for repeatedly lost tasks.
Docstring Coverage	✅ Passed	No functions found in the changed files to evaluate docstring coverage. Skipping docstring coverage check.

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch feat/worker-loss-cap-main

---

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.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands.}

[coderabbitai]

Actionable comments posted: 2

🧹 Nitpick comments (6)

secator/celery.py (2)

341-344: 🎯 Functional Correctness | 🔵 Trivial | ⚡ Quick win

Gate the cap on task_reject_on_worker_lost too.

The cap only targets worker-loss redeliveries; requiring both late acks and reject-on-worker-lost keeps the runtime behavior aligned with the feature’s activation contract.

Proposed fix

-		if CONFIG.celery.task_max_retries != -1 and CONFIG.celery.task_acks_late:
+		if (
+			CONFIG.celery.task_max_retries != -1
+			and CONFIG.celery.task_acks_late
+			and CONFIG.celery.task_reject_on_worker_lost
+		):

🤖 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 341 - 344, The retry cap in the worker-loss
redelivery path is only guarded by `task_acks_late`, but it should also require
`task_reject_on_worker_lost` to match the feature contract. Update the condition
in the `celery.py` task handling logic around `bump_worker_loss_count`,
`worker_loss_retries_exhausted`, and `abandon_task` so the cap runs only when
both `CONFIG.celery.task_acks_late` and
`CONFIG.celery.task_reject_on_worker_lost` are enabled, keeping the worker-loss
redelivery behavior aligned with the intended activation settings.

---

211-214: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win

Keep TTL failures diagnosable.

The best-effort ignore is fine for unsupported expire, but unexpected backend errors should be debug-logged instead of silently swallowed.

Proposed fix

 	try:
 		backend.expire(key, CONFIG.celery.result_expires)
-	except Exception:
+	except (AttributeError, NotImplementedError):
 		pass
+	except Exception as e:  # noqa: BLE001
+		debug(f'worker-loss expire failed for {key}: {e}', sub='celery.state')

🤖 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 211 - 214, The backend TTL update in the
expire call is swallowing all failures silently, which hides unexpected backend
issues. Update the try/except around backend.expire in the celery
result-expiration path to keep the best-effort behavior for unsupported expire,
but catch unexpected exceptions and emit a debug log with the error details
instead of using a bare pass. Use the existing celery/backend expiration flow
and the surrounding CONFIG.celery.result_expires context to locate the fix.

Source: Linters/SAST tools

tests/unit/test_celery.py (4)

360-365: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value

Blind except Exception: pass in cleanup silently hides delete failures.

Ruff flags S110/BLE001 here. Since this is test-cleanup code, low risk, but a debug log would help diagnose flaky cross-test key collisions instead of silently swallowing errors.

🔧 Suggested fix

 		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:
+				print(f'cleanup failed: {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 360 - 365, The cleanup block in the
test teardown is swallowing all failures with a blind exception handler, which
hides backend delete issues and makes flaky key-collision problems hard to
diagnose. Update the cleanup around app.backend.delete and
app.backend.get_key_for_task in the test_celery cleanup path to catch the
exception explicitly and emit a debug log with the key information before
continuing, keeping the teardown non-fatal but observable.

Source: Linters/SAST tools

---

349-349: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win

Local app re-imports shadow the module-level import (flake8 F811).

Static analysis flags redefinition of app at each of these lines since it's already imported at module level. Per the project's flake8 config (max-line-length=120, ignoring only W191/E101/E128/E265/W605), F811 is not suppressed and should be fixed.

🔧 Suggested fix

 	def test_bump_worker_loss_count_get_set_fallback(self):
 		"""Counter increments via the generic get/set fallback (any KV backend, e.g. filesystem)."""
-		from secator.celery import app, bump_worker_loss_count
+		from secator.celery import bump_worker_loss_count

Apply the analogous change at lines 370, 379, and 415 (drop the redundant app import, relying on the module-level import instead).

Also applies to: 370-370, 379-379, 415-415

🤖 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 349, Remove the redundant local re-imports
of app in the celery test cases to avoid shadowing the module-level import and
triggering flake8 F811. Update the affected test blocks in test_celery by
keeping the existing module-level app import and only importing the additional
symbols needed there, including the sections around bump_worker_loss_count and
the analogous cases at the other referenced spots.

Sources: Coding guidelines, Linters/SAST tools

---

385-391: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win

Bare return silently skips the test instead of reporting it as skipped.

If httpx not in TEST_TASKS, the test method returns early without any assertion, making the test appear to "pass" in CI while never actually validating abandon_task's failure-result behavior. This hides missing coverage.

🔧 Suggested fix

 		from secator.tasks import httpx
 		from secator.celery import abandon_task
-		if httpx not in TEST_TASKS:
-			return
+		if httpx not in TEST_TASKS:
+			self.skipTest('httpx task not available in TEST_TASKS')

🤖 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 385 - 391, The test in
test_abandon_task_returns_failure_error should not use a bare early return when
httpx is absent from TEST_TASKS, because that makes the case look like a passing
test instead of a skipped one. Update the test method to explicitly mark it as
skipped using the test framework’s skip mechanism before exercising
abandon_task, so the intent is visible in CI while keeping the failure-result
behavior covered when httpx is available.

---

381-382: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win

Over-indented continuation line (flake8 E127).

Line 382's continuation for the with ... , \ statement is over-indented for visual indent. E127 is not in the project's flake8 ignore list.

🔧 Suggested 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):

🤖 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 381 - 382, The `with
patch.object(...)` statement in `test_celery.py` has an over-indented
continuation line that triggers flake8 E127. Adjust the indentation of the
continuation after the backslash in the `with` block so it matches the expected
visual indent style, keeping the line aligned consistently with the surrounding
test code in `test_celery.py`.

Sources: Coding guidelines, Linters/SAST tools

🤖 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 336-344: The early abandon path in the task redelivery cap check
returns an abandoned task without the request context attached, so ensure
`context` is added to `opts` before calling `abandon_task` in the `worker-loss
redelivery cap` block inside `celery.py`. Update the logic around
`bump_worker_loss_count`, `worker_loss_retries_exhausted`, and the
`abandon_task` call so the abandoned task always includes `celery_id`,
`worker_name`, and `routing_key` from `self.request` even when `opts` did not
already contain `context`.

In `@secator/config.py`:
- Line 77: The task_max_retries config in Config currently allows invalid values
below -1, which can cause run_command and worker_loss_retries_exhausted to
abandon tasks immediately; update the Config field definition for
task_max_retries to enforce a minimum of -1 using the existing Pydantic Field
pattern used elsewhere in secator/config.py.

---

Nitpick comments:
In `@secator/celery.py`:
- Around line 341-344: The retry cap in the worker-loss redelivery path is only
guarded by `task_acks_late`, but it should also require
`task_reject_on_worker_lost` to match the feature contract. Update the condition
in the `celery.py` task handling logic around `bump_worker_loss_count`,
`worker_loss_retries_exhausted`, and `abandon_task` so the cap runs only when
both `CONFIG.celery.task_acks_late` and
`CONFIG.celery.task_reject_on_worker_lost` are enabled, keeping the worker-loss
redelivery behavior aligned with the intended activation settings.
- Around line 211-214: The backend TTL update in the expire call is swallowing
all failures silently, which hides unexpected backend issues. Update the
try/except around backend.expire in the celery result-expiration path to keep
the best-effort behavior for unsupported expire, but catch unexpected exceptions
and emit a debug log with the error details instead of using a bare pass. Use
the existing celery/backend expiration flow and the surrounding
CONFIG.celery.result_expires context to locate the fix.

In `@tests/unit/test_celery.py`:
- Around line 360-365: The cleanup block in the test teardown is swallowing all
failures with a blind exception handler, which hides backend delete issues and
makes flaky key-collision problems hard to diagnose. Update the cleanup around
app.backend.delete and app.backend.get_key_for_task in the test_celery cleanup
path to catch the exception explicitly and emit a debug log with the key
information before continuing, keeping the teardown non-fatal but observable.
- Line 349: Remove the redundant local re-imports of app in the celery test
cases to avoid shadowing the module-level import and triggering flake8 F811.
Update the affected test blocks in test_celery by keeping the existing
module-level app import and only importing the additional symbols needed there,
including the sections around bump_worker_loss_count and the analogous cases at
the other referenced spots.
- Around line 385-391: The test in test_abandon_task_returns_failure_error
should not use a bare early return when httpx is absent from TEST_TASKS, because
that makes the case look like a passing test instead of a skipped one. Update
the test method to explicitly mark it as skipped using the test framework’s skip
mechanism before exercising abandon_task, so the intent is visible in CI while
keeping the failure-result behavior covered when httpx is available.
- Around line 381-382: The `with patch.object(...)` statement in
`test_celery.py` has an over-indented continuation line that triggers flake8
E127. Adjust the indentation of the continuation after the backslash in the
`with` block so it matches the expected visual indent style, keeping the line
aligned consistently with the surrounding test code in `test_celery.py`.

🪄 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: 864e2469-9da8-4b62-8df1-f647be174aa1

📥 Commits

Reviewing files that changed from the base of the PR and between 0203f02 and 536b1f7.

📒 Files selected for processing (3)

• secator/celery.py
• secator/config.py
• tests/unit/test_celery.py

[ocervell]

Addressed the CodeRabbit findings (commit pushed):

• Attach context before abandon — opts['context'] = context is now set before the early abandon_task return, so an abandoned task keeps celery_id/worker_name/routing_key even when opts had no context.
• Gate on task_reject_on_worker_lost — the cap now requires task_acks_late and task_reject_on_worker_lost (the setting that actually re-queues on abrupt worker death).
• task_max_retries validation — Field(default=-1, ge=-1) so values < -1 can't enable the cap and abandon on first delivery.
• Expire diagnosability — _expire_worker_loss_key now only swallows AttributeError/NotImplementedError; other backend errors are debug-logged.

tests/unit/test_celery.py 23/23, flake8 clean.