From e4975329ead465366c0800ced2824b170157c55d Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Thu, 16 Jul 2026 01:17:16 +0530 Subject: [PATCH 1/3] feat(scheduler): cap retry rate per task with a retry budget Nothing bounded aggregate retries: max_retries is a per-job budget, and the circuit breaker trips on hard failure, not on retry rate. A dependency that fails every job of a task retries without limit. @queue.task(retry_budget="100/m") caps how fast a task may retry across all its jobs; once spent, failures dead-letter with metadata recording why. Reuses the token-bucket rate limiter, which already takes a free-form key and is atomic on all three backends, so this needs no new storage. Keyed per namespace, unlike the circuit breaker, so tenants sharing a database cannot spend each other's budget. The token is taken only when a retry would actually happen, so a job at its retry ceiling cannot drain the budget for its siblings. --- crates/taskito-core/src/scheduler/mod.rs | 178 ++++++++++++++++++ .../src/scheduler/result_handler.rs | 40 +++- crates/taskito-java/src/worker.rs | 2 + .../taskito-node/src/convert/task_config.rs | 2 + crates/taskito-python/src/py_config.rs | 7 +- crates/taskito-python/src/py_queue/worker.rs | 13 ++ sdks/python/taskito/_taskito.pyi | 2 + sdks/python/taskito/mixins/decorators.py | 15 +- 8 files changed, 245 insertions(+), 14 deletions(-) diff --git a/crates/taskito-core/src/scheduler/mod.rs b/crates/taskito-core/src/scheduler/mod.rs index d9355dc8..105cdb0e 100644 --- a/crates/taskito-core/src/scheduler/mod.rs +++ b/crates/taskito-core/src/scheduler/mod.rs @@ -145,6 +145,11 @@ pub struct TaskConfig { pub retry_policy: RetryPolicy, pub rate_limit: Option, pub circuit_breaker: Option, + /// Cap on how fast this task may *retry*, across every job of the task. + /// Nothing else catches sustained retry storms: the circuit breaker trips on + /// hard failure, not on aggregate retry rate, and per-job `max_retries` is a + /// per-job budget, not a rate. Exhausting it dead-letters instead of retrying. + pub retry_budget: Option, pub max_concurrent: Option, /// Cap on this task's share of *this worker's* in-flight slots, so one slow /// task cannot occupy the whole pool and starve every other task. Complements @@ -327,6 +332,35 @@ impl Scheduler { .count_for(task_name) } + /// Take a token from this task's retry budget. `true` when the retry may go + /// ahead — including when no budget is configured, which is the default. + /// + /// Keyed per namespace so tenants sharing a database cannot spend each + /// other's budget. The bucket is created on first use, so nothing needs + /// registering. A storage error must not strand the job, so it fails open: + /// the per-job `max_retries` still bounds the retry. + fn retry_budget_allows(&self, task_name: &str) -> bool { + let Some(config) = self + .task_configs + .get(task_name) + .and_then(|c| c.retry_budget.as_ref()) + else { + return true; + }; + let key = format!( + "retry::{}::{}", + self.namespace.as_deref().unwrap_or(""), + task_name + ); + match self.rate_limiter.try_acquire(&key, config) { + Ok(allowed) => allowed, + Err(e) => { + error!("retry budget check failed for {task_name}: {e}"); + true + } + } + } + /// Release a job's in-flight slot when it finishes and wake the poller to /// refill. Only ids this scheduler dispatched are tracked, so results for /// recovered or foreign jobs (e.g. from maintenance) are a harmless no-op. @@ -759,6 +793,7 @@ mod tests { }, rate_limit: None, circuit_breaker: None, + retry_budget: None, max_concurrent: None, max_in_flight_per_task: None, }, @@ -800,6 +835,7 @@ mod tests { }, rate_limit: None, circuit_breaker: None, + retry_budget: None, max_concurrent: None, max_in_flight_per_task: None, }, @@ -902,6 +938,7 @@ mod tests { refill_rate: 0.0, }), circuit_breaker: None, + retry_budget: None, max_concurrent: None, max_in_flight_per_task: None, }, @@ -947,6 +984,7 @@ mod tests { retry_policy: RetryPolicy::default(), rate_limit: None, circuit_breaker: Some(cb_config), + retry_budget: None, max_concurrent: None, max_in_flight_per_task: None, }, @@ -986,6 +1024,7 @@ mod tests { retry_policy: RetryPolicy::default(), rate_limit: None, circuit_breaker: None, + retry_budget: None, max_concurrent: Some(2), max_in_flight_per_task: None, }, @@ -1055,6 +1094,7 @@ mod tests { retry_policy: RetryPolicy::default(), rate_limit: None, circuit_breaker: None, + retry_budget: None, max_concurrent: Some(2), max_in_flight_per_task: None, }, @@ -1156,6 +1196,7 @@ mod tests { retry_policy: RetryPolicy::default(), rate_limit: None, circuit_breaker: None, + retry_budget: None, max_concurrent: None, max_in_flight_per_task: Some(per_task), }, @@ -1277,6 +1318,7 @@ mod tests { retry_policy: RetryPolicy::default(), rate_limit: None, circuit_breaker: None, + retry_budget: None, max_concurrent: Some(1), max_in_flight_per_task: None, }, @@ -1442,6 +1484,7 @@ mod tests { }, rate_limit: None, circuit_breaker: None, + retry_budget: None, max_concurrent: None, max_in_flight_per_task: None, }, @@ -1479,6 +1522,7 @@ mod tests { }, rate_limit: None, circuit_breaker: None, + retry_budget: None, max_concurrent: None, max_in_flight_per_task: None, } @@ -1521,6 +1565,140 @@ mod tests { /// A scheduler must never orphan its OWN in-flight jobs (self-rescue guard), /// even before its first heartbeat row exists. + /// A task whose retries are capped at `budget` per minute. + fn budgeted_scheduler(task_name: &str, budget: &str) -> Scheduler { + let mut scheduler = test_scheduler(); + let mut config = retry_task_config(5); + config.retry_budget = Some(RateLimitConfig::parse(budget).expect("valid rate")); + scheduler.register_task(task_name.to_string(), config); + scheduler + } + + /// Fail `job_id` once and report what the scheduler decided. + fn fail_once(scheduler: &Scheduler, job_id: &str, task_name: &str) -> ResultOutcome { + scheduler + .handle_result(JobResult::Failure { + job_id: job_id.to_string(), + error: "boom".to_string(), + retry_count: 0, + max_retries: 5, + task_name: task_name.to_string(), + wall_time_ns: 1, + should_retry: true, + timed_out: false, + }) + .unwrap() + } + + #[test] + fn test_retry_budget_dead_letters_once_exhausted() { + // S18: the per-job max_retries is not a rate cap, so a broken dependency + // can retry forever across jobs. Two tokens, three failures: the third + // must dead-letter instead of retrying. + let scheduler = budgeted_scheduler("budgeted", "2/m"); + + let mut outcomes = Vec::new(); + for _ in 0..3 { + let job = scheduler.storage.enqueue(make_job("budgeted")).unwrap(); + outcomes.push(fail_once(&scheduler, &job.id, "budgeted")); + } + + assert!( + matches!(outcomes[0], ResultOutcome::Retry { .. }), + "first failure is within budget" + ); + assert!( + matches!(outcomes[1], ResultOutcome::Retry { .. }), + "second failure is within budget" + ); + assert!( + matches!(outcomes[2], ResultOutcome::DeadLettered { .. }), + "third failure exceeds the budget and must dead-letter, got {:?}", + outcomes[2] + ); + } + + #[test] + fn test_retry_budget_marks_why_it_dead_lettered() { + // ResultOutcome can't say why, so the DLQ metadata is what distinguishes + // a budget kill from ordinary retry exhaustion. + let scheduler = budgeted_scheduler("marked", "1/m"); + for _ in 0..2 { + let job = scheduler.storage.enqueue(make_job("marked")).unwrap(); + fail_once(&scheduler, &job.id, "marked"); + } + + let dead = scheduler.storage.list_dead(10, 0).unwrap(); + assert_eq!(dead.len(), 1); + assert_eq!( + dead[0].metadata.as_deref(), + Some(super::result_handler::RETRY_BUDGET_EXHAUSTED) + ); + } + + #[test] + fn test_retry_budget_is_per_task() { + // One task draining its budget must not stop another from retrying. + let mut scheduler = budgeted_scheduler("greedy", "1/m"); + scheduler.register_task("innocent".to_string(), retry_task_config(5)); + + for _ in 0..2 { + let job = scheduler.storage.enqueue(make_job("greedy")).unwrap(); + fail_once(&scheduler, &job.id, "greedy"); + } + + let job = scheduler.storage.enqueue(make_job("innocent")).unwrap(); + let outcome = fail_once(&scheduler, &job.id, "innocent"); + assert!( + matches!(outcome, ResultOutcome::Retry { .. }), + "an unbudgeted task must be unaffected, got {outcome:?}" + ); + } + + #[test] + fn test_no_retry_budget_configured_never_blocks() { + // The budget is opt-in; the default must keep retrying to max_retries. + let mut scheduler = test_scheduler(); + scheduler.register_task("unbudgeted".to_string(), retry_task_config(5)); + for _ in 0..5 { + let job = scheduler.storage.enqueue(make_job("unbudgeted")).unwrap(); + let outcome = fail_once(&scheduler, &job.id, "unbudgeted"); + assert!(matches!(outcome, ResultOutcome::Retry { .. })); + } + } + + #[test] + fn test_retry_budget_not_spent_by_jobs_that_cannot_retry() { + // A job at its retry ceiling was never going to retry, so it must not + // take a token — otherwise it drains the budget for its siblings. + let scheduler = budgeted_scheduler("ceiling", "1/m"); + + let exhausted = scheduler.storage.enqueue(make_job("ceiling")).unwrap(); + let outcome = scheduler + .handle_result(JobResult::Failure { + job_id: exhausted.id.clone(), + error: "boom".to_string(), + retry_count: 3, + max_retries: 3, // already at the ceiling + task_name: "ceiling".to_string(), + wall_time_ns: 1, + should_retry: true, + timed_out: false, + }) + .unwrap(); + assert!(matches!(outcome, ResultOutcome::DeadLettered { .. })); + + // The single token must still be there for a job that can retry. + let fresh = scheduler.storage.enqueue(make_job("ceiling")).unwrap(); + assert!( + matches!( + fail_once(&scheduler, &fresh.id, "ceiling"), + ResultOutcome::Retry { .. } + ), + "retry-exhausted job must not spend a budget token" + ); + } + #[test] fn test_recover_orphaned_jobs_skips_own_claims() { let mut scheduler = test_scheduler(); diff --git a/crates/taskito-core/src/scheduler/result_handler.rs b/crates/taskito-core/src/scheduler/result_handler.rs index e5bf24fb..734d025a 100644 --- a/crates/taskito-core/src/scheduler/result_handler.rs +++ b/crates/taskito-core/src/scheduler/result_handler.rs @@ -6,6 +6,11 @@ use crate::storage::Storage; use super::{JobResult, ResultOutcome, Scheduler}; +/// Dead-letter metadata marking a job the retry budget refused. `ResultOutcome` +/// has no room to say *why* a job was dead-lettered, so this is what tells a +/// budget kill apart from ordinary retry exhaustion when reading the DLQ. +pub const RETRY_BUDGET_EXHAUSTED: &str = "retry_budget_exhausted"; + impl Scheduler { /// Handle a completed or failed job result from a worker. /// @@ -62,19 +67,20 @@ impl Scheduler { let job = self.storage.get_job(&job_id)?; let queue = job.as_ref().map(|j| j.queue.clone()).unwrap_or_default(); - let move_to_dlq = |job: Option<&crate::job::Job>| -> Result<()> { - match job { - Some(j) => self.dlq.move_to_dlq(j, &error, None), - None => { - warn!("job {job_id} disappeared before DLQ move"); - Ok(()) + let move_to_dlq = + |job: Option<&crate::job::Job>, metadata: Option<&str>| -> Result<()> { + match job { + Some(j) => self.dlq.move_to_dlq(j, &error, metadata), + None => { + warn!("job {job_id} disappeared before DLQ move"); + Ok(()) + } } - } - }; + }; // If should_retry is false (exception filtering), skip straight to DLQ if !should_retry { - move_to_dlq(job.as_ref())?; + move_to_dlq(job.as_ref(), None)?; return Ok(ResultOutcome::DeadLettered { job_id, task_name, @@ -97,6 +103,20 @@ impl Scheduler { let effective_max = max_retries; if retry_count < effective_max { + // Checked here, not before the per-job budget: a job that was + // never going to retry must not spend a token, or a task at + // its retry ceiling would drain the budget for its siblings. + if !self.retry_budget_allows(&task_name) { + warn!("retry budget exhausted for {task_name}; dead-lettering {job_id}"); + move_to_dlq(job.as_ref(), Some(RETRY_BUDGET_EXHAUSTED))?; + return Ok(ResultOutcome::DeadLettered { + job_id, + task_name, + queue, + error, + timed_out, + }); + } let next_at = policy.next_retry_at(retry_count); self.storage.retry(&job_id, next_at)?; #[cfg(feature = "push-dispatch")] @@ -110,7 +130,7 @@ impl Scheduler { timed_out, }) } else { - move_to_dlq(job.as_ref())?; + move_to_dlq(job.as_ref(), None)?; Ok(ResultOutcome::DeadLettered { job_id, task_name, diff --git a/crates/taskito-java/src/worker.rs b/crates/taskito-java/src/worker.rs index 1b67c709..d5f8d69b 100644 --- a/crates/taskito-java/src/worker.rs +++ b/crates/taskito-java/src/worker.rs @@ -277,6 +277,8 @@ fn register_task_policies(scheduler: &mut Scheduler, configs: Option Result { .as_ref() .map(circuit_breaker_config) .transpose()?, + // Not surfaced on this SDK yet; retries stay bounded per job. + retry_budget: None, max_concurrent: input.max_concurrent, // Not surfaced on this SDK yet; the whole pool stays available. max_in_flight_per_task: None, diff --git a/crates/taskito-python/src/py_config.rs b/crates/taskito-python/src/py_config.rs index 0494ad8f..4e3d24cb 100644 --- a/crates/taskito-python/src/py_config.rs +++ b/crates/taskito-python/src/py_config.rs @@ -36,13 +36,16 @@ pub struct PyTaskConfig { pub circuit_breaker_half_open_success_rate: Option, #[pyo3(get, set)] pub max_in_flight_per_task: Option, + /// Retry-rate cap, same syntax as `rate_limit` (e.g. "100/m"). + #[pyo3(get, set)] + pub retry_budget: Option, } #[pymethods] #[allow(clippy::too_many_arguments)] impl PyTaskConfig { #[new] - #[pyo3(signature = (name, max_retries=3, retry_backoff=1.0, timeout=300, priority=0, rate_limit=None, queue="default".to_string(), circuit_breaker_threshold=None, circuit_breaker_window=None, circuit_breaker_cooldown=None, retry_delays=None, max_retry_delay=None, max_concurrent=None, circuit_breaker_half_open_probes=None, circuit_breaker_half_open_success_rate=None, max_in_flight_per_task=None))] + #[pyo3(signature = (name, max_retries=3, retry_backoff=1.0, timeout=300, priority=0, rate_limit=None, queue="default".to_string(), circuit_breaker_threshold=None, circuit_breaker_window=None, circuit_breaker_cooldown=None, retry_delays=None, max_retry_delay=None, max_concurrent=None, circuit_breaker_half_open_probes=None, circuit_breaker_half_open_success_rate=None, max_in_flight_per_task=None, retry_budget=None))] pub fn new( name: String, max_retries: i32, @@ -60,6 +63,7 @@ impl PyTaskConfig { circuit_breaker_half_open_probes: Option, circuit_breaker_half_open_success_rate: Option, max_in_flight_per_task: Option, + retry_budget: Option, ) -> Self { Self { name, @@ -78,6 +82,7 @@ impl PyTaskConfig { circuit_breaker_half_open_probes, circuit_breaker_half_open_success_rate, max_in_flight_per_task, + retry_budget, } } } diff --git a/crates/taskito-python/src/py_queue/worker.rs b/crates/taskito-python/src/py_queue/worker.rs index 91e14ef2..6b8fa49a 100644 --- a/crates/taskito-python/src/py_queue/worker.rs +++ b/crates/taskito-python/src/py_queue/worker.rs @@ -365,6 +365,18 @@ impl PyQueue { .rate_limit .as_ref() .and_then(|s| RateLimitConfig::parse(s)); + // Same "100/m" syntax as rate_limit, so it parses the same way. An + // unparseable value would silently disable the cap, so reject it here + // rather than let a typo look like it took effect. + let retry_budget = match tc.retry_budget.as_ref() { + Some(s) => Some(RateLimitConfig::parse(s).ok_or_else(|| { + pyo3::exceptions::PyValueError::new_err(format!( + "invalid retry_budget {s:?} for task {}: expected a rate like \"100/m\"", + tc.name + )) + })?), + None => None, + }; let circuit_breaker = tc.circuit_breaker_threshold .map(|threshold| CircuitBreakerConfig { @@ -382,6 +394,7 @@ impl PyQueue { retry_policy, rate_limit, circuit_breaker, + retry_budget, max_concurrent: tc.max_concurrent, max_in_flight_per_task: tc.max_in_flight_per_task.map(|n| n.max(1) as usize), }, diff --git a/sdks/python/taskito/_taskito.pyi b/sdks/python/taskito/_taskito.pyi index 12bff94a..0f3e2e95 100644 --- a/sdks/python/taskito/_taskito.pyi +++ b/sdks/python/taskito/_taskito.pyi @@ -25,6 +25,7 @@ class PyTaskConfig: circuit_breaker_half_open_probes: int | None circuit_breaker_half_open_success_rate: float | None max_in_flight_per_task: int | None + retry_budget: str | None def __init__( self, @@ -44,6 +45,7 @@ class PyTaskConfig: circuit_breaker_half_open_probes: int | None = None, circuit_breaker_half_open_success_rate: float | None = None, max_in_flight_per_task: int | None = None, + retry_budget: str | None = None, ) -> None: ... class PyJob: diff --git a/sdks/python/taskito/mixins/decorators.py b/sdks/python/taskito/mixins/decorators.py index b26340b0..4dce800a 100644 --- a/sdks/python/taskito/mixins/decorators.py +++ b/sdks/python/taskito/mixins/decorators.py @@ -385,10 +385,11 @@ def task( on_false: str = "defer", predicate_extras: dict[str, Any] | None = None, default_defer_seconds: float = 60.0, - # Appended, not slotted next to `max_concurrent`: this signature is not - # keyword-only, so inserting mid-list would silently rebind the positional - # arguments after it. + # Appended, not slotted next to their related options: this signature is + # not keyword-only, so inserting mid-list would silently rebind the + # positional arguments after it. max_in_flight_per_task: int | None = None, + retry_budget: str | None = None, ) -> Callable[[Callable[..., Any]], TaskWrapper]: """Decorator to register a function as a background task. @@ -430,6 +431,13 @@ def task( starve the others. Unlike ``max_concurrent`` (a cluster-wide cap that costs a database read), this is in-process and free. ``None`` lets the task use the whole pool. + retry_budget: Cap on how fast this task may *retry*, across all of its + jobs — same syntax as ``rate_limit`` (e.g. ``"100/m"``). Once the + budget is spent, further failures dead-letter instead of retrying, + which stops a broken dependency turning into a retry storm. + Distinct from ``max_retries``, which bounds one job rather than the + rate, and from ``circuit_breaker``, which trips on hard failure + rather than aggregate retry rate. ``None`` means no cap. compensates: Optional reference to a task that compensates this one. When this task runs as part of a workflow saga and a later step fails, the framework enqueues the compensation @@ -633,6 +641,7 @@ def decorator(fn: Callable) -> TaskWrapper: circuit_breaker_half_open_probes=cb_half_open_probes, circuit_breaker_half_open_success_rate=cb_half_open_success_rate, max_in_flight_per_task=max_in_flight_per_task, + retry_budget=retry_budget, ) self._task_configs.append(config) From d1d906183ca9327b62f646ddc4209dec8af8af9c Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Thu, 16 Jul 2026 01:17:32 +0530 Subject: [PATCH 2/3] test(retry): cover the retry budget end to end Verified both new tests fail with the gate removed: the storm test then sees only retry-exhausted dead letters (metadata None) rather than budget kills. --- sdks/python/tests/core/test_retry.py | 58 ++++++++++++++++++++++++++++ 1 file changed, 58 insertions(+) diff --git a/sdks/python/tests/core/test_retry.py b/sdks/python/tests/core/test_retry.py index 00af8c50..1fb171e4 100644 --- a/sdks/python/tests/core/test_retry.py +++ b/sdks/python/tests/core/test_retry.py @@ -85,3 +85,61 @@ def fail_once() -> None: new_id = queue.retry_dead(dead[0]["id"]) assert new_id is not None assert len(new_id) > 0 + + +def test_retry_budget_dead_letters_a_retry_storm(queue: Queue, poll_until: PollUntil) -> None: + """retry_budget caps the retry *rate* across a task's jobs, not per job. + + max_retries bounds one job; a broken dependency failing every job still + retries without limit. Once the budget is spent, failures dead-letter. + """ + attempts = 0 + + @queue.task(max_retries=5, retry_backoff=0.01, retry_budget="2/m") + def always_fails() -> None: + nonlocal attempts + attempts += 1 + raise RuntimeError("dependency down") + + for _ in range(4): + always_fails.delay() + + thread = threading.Thread(target=queue.run_worker, daemon=True) + thread.start() + try: + poll_until( + lambda: len(queue.dead_letters(limit=10)) >= 2, + timeout=30, + message="budget-exhausted jobs never reached the DLQ", + ) + finally: + queue.shutdown() + thread.join(timeout=5) + + dead = queue.dead_letters(limit=10) + budget_killed = [d for d in dead if d.get("metadata") == "retry_budget_exhausted"] + assert budget_killed, f"no job dead-lettered by the budget; DLQ={dead}" + + +def test_retry_budget_rejects_an_unparseable_rate(queue: Queue) -> None: + """A typo must not silently disable the cap.""" + + @queue.task(retry_budget="not-a-rate") + def bad_budget() -> None: + pass + + thread_error: list[BaseException] = [] + + def _run() -> None: + try: + queue.run_worker() + except BaseException as exc: + thread_error.append(exc) + + thread = threading.Thread(target=_run, daemon=True) + thread.start() + thread.join(timeout=10) + queue.shutdown() + + assert thread_error, "an invalid retry_budget must be rejected, not ignored" + assert "retry_budget" in str(thread_error[0]) From b1dce478619f093ab73203c6261112620de934f7 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Thu, 16 Jul 2026 01:17:39 +0530 Subject: [PATCH 3/3] docs: document retry_budget and max_in_flight_per_task Both are public @queue.task options missing from the parameter reference. --- docs/content/docs/python/api-reference/queue/index.mdx | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/docs/content/docs/python/api-reference/queue/index.mdx b/docs/content/docs/python/api-reference/queue/index.mdx index 0d3b9c91..c16e5d0c 100644 --- a/docs/content/docs/python/api-reference/queue/index.mdx +++ b/docs/content/docs/python/api-reference/queue/index.mdx @@ -93,6 +93,8 @@ Queue( batch: bool | dict | None = None, predicate: Predicate | Callable | None = None, on_false: str = "defer", + max_in_flight_per_task: int | None = None, + retry_budget: str | None = None, ) -> TaskWrapper ``` @@ -115,7 +117,9 @@ Register a function as a background task. Returns a [`TaskWrapper`](/python/api- | `middleware` | `list[TaskMiddleware] \| None` | `None` | Per-task middleware, applied in addition to queue-level middleware. | | `inject` | `list[str] \| None` | `None` | Resource names to inject as keyword arguments. See [Resource System](/python/guides/resources). | | `serializer` | `Serializer \| None` | `None` | Per-task serializer override. Falls back to queue-level serializer. | -| `max_concurrent` | `int \| None` | `None` | Max concurrent running instances. `None` = no limit. | +| `max_concurrent` | `int \| None` | `None` | Max concurrent running instances, across the cluster. `None` = no limit. | +| `max_in_flight_per_task` | `int \| None` | `None` | Cap on this task's share of a *single worker's* dispatch slots, so one slow task can't occupy the whole pool and starve the others. In-process and free, unlike `max_concurrent`, which is cluster-wide and costs a database read. `None` = the task may use the whole pool. | +| `retry_budget` | `str \| None` | `None` | Cap on how fast this task may **retry**, across all of its jobs — same syntax as `rate_limit`, e.g. `"100/m"`. Once spent, further failures dead-letter instead of retrying (DLQ `metadata` is set to `retry_budget_exhausted`). Distinct from `max_retries`, which bounds one job rather than the rate, and from `circuit_breaker`, which trips on hard failure rather than aggregate retry rate. `None` = no cap. | | `idempotent` | `bool` | `False` | Auto-derive a dedup key from `sha256(task_name\|payload)` so duplicate `.delay()`/`.apply_async()` calls while a job is pending or running return the same job ID. | | `compensates` | `TaskWrapper \| str \| None` | `None` | Saga compensator for this task, enqueued in reverse order if a later workflow step fails. See [Sagas](/python/guides/workflows/saga). | | `batch` | `bool \| dict \| None` | `None` | Enable producer-side batching (`True` for defaults, or a dict with `max_size`/`max_wait_ms`). See [Batching](/python/guides/core/batching). |