fix: bound RetryTaskException so permanently-failing tasks stop looping

[adhikjoshi]

Problem

RetryTaskException is re-queued unconditionally, with no attempt counter:

except RetryTaskException as e:
    logger.warning(f"Task {task.task_name} requested retry: {e}")
    new_task_dict = task.to_dict()
    new_task_dict["payload"] = task.original_payload
    self.enqueue_delayed_task(new_task_dict, delay_seconds=self.delay_seconds)

A task whose failure is permanent therefore never stops. It re-enters the queue every delay_seconds, forever.

Impact measured in production

On one image worker, over a single 3,238-line log window:

Tasks started	120
Requested retry	119
Distinct task ids involved	47
Max retries seen on one task	9 (within that window alone)

Every failure was a source image that no longer exists — HTTP 404 from an ephemeral delivery URL, or HTTP 403 from a CloudFront signature that had expired 3–4 days earlier. None could ever succeed, and together they were consuming essentially the whole worker capacity of the container.

Fix

Bound the loop.

• The counter rides on the payload so it survives the round trip through Redis.
• It uses a reserved _retry_attempts key, so it cannot collide with the caller-facing retries budget that the TaskProcessingError path already decrements (base.py ~864–877).
• Once the budget is spent, the task goes through the normal failure path — final state stored, error logged, on_error middleware, webhook — instead of being silently re-queued.
• Configurable via retry_task_max_attempts, default 3, so genuinely transient retries (e.g. a model that briefly fails to load) behave as before.

Tests

tests/test_retry_budget.py covers:

• the budget being spent, and no further re-queue afterwards
• the reserved counter leaving the caller's own retries value untouched
• a zero budget failing immediately
• the default being finite (the old behaviour was unbounded)

66 passed   (with fix)
62 passed   (baseline, new file excluded)

No regressions; the 4 new tests are the difference.

Note

This caps the blast radius for every task type. The matching upstream fix — not classifying a permanent 4xx download failure as retryable in the first place — is handled separately in dreambooth-server.

---

^{Need help on this PR? Tag @codesmith-bot with what you need. Autofix is enabled.}