From 785584ea02fe2793c83db60ed3b5537924cb0fb6 Mon Sep 17 00:00:00 2001
From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com>
Date: Thu, 16 Jul 2026 10:42:28 +0530
Subject: [PATCH 01/13] fix(node): bound in-flight jobs to a concurrency option
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The dispatcher spawned every job it was handed and immediately took the next,
so the job channel drained as fast as the scheduler filled it. A channel that
never fills applies no backpressure, and this SDK never set max_in_flight —
which Python and Java both do — so nothing capped in-flight work. The worker
claimed jobs it could not run, stranding them Running while peers sharing the
database skipped them.

Add a `concurrency` option, separate from `channelCapacity`, mirroring Java:
it caps what the scheduler claims and what the dispatcher runs. The dispatcher
permit is the only bound on the mesh path, where stolen jobs never reach this
scheduler's in-flight accounting.

Opt-in — unset stays unbounded, so no existing worker silently acquires a cap.
channelCapacity's doc claimed to be the in-flight bound and never was; it now
says what it does.
---
 crates/taskito-node/src/config.rs          |  4 ++
 crates/taskito-node/src/dispatcher.rs      | 40 +++++++++++--
 crates/taskito-node/src/worker.rs          | 12 +++-
 sdks/node/src/types.ts                     | 12 +++-
 sdks/node/src/worker.ts                    |  1 +
 sdks/node/test/worker/backpressure.test.ts | 65 ++++++++++++++++++++++
 6 files changed, 127 insertions(+), 7 deletions(-)
 create mode 100644 sdks/node/test/worker/backpressure.test.ts

diff --git a/crates/taskito-node/src/config.rs b/crates/taskito-node/src/config.rs
index a185fafb..9c2a8b1d 100644
--- a/crates/taskito-node/src/config.rs
+++ b/crates/taskito-node/src/config.rs
@@ -160,6 +160,10 @@ pub struct MeshWorkerConfig {
 pub struct WorkerOptions {
     pub queues: Option<Vec<String>>,
     pub channel_capacity: Option<u32>,
+    /// Jobs this worker runs at once. Unset leaves in-flight work unbounded,
+    /// which is the historical behaviour — a worker then claims jobs it cannot
+    /// run yet, stranding them Running and starving peers on the same database.
+    pub concurrency: Option<u32>,
     /// Jobs claimed per scheduler poll (default 1).
     pub batch_size: Option<u32>,
     pub task_configs: Option<Vec<TaskConfigInput>>,
diff --git a/crates/taskito-node/src/dispatcher.rs b/crates/taskito-node/src/dispatcher.rs
index 42f8b2c1..39aec9b7 100644
--- a/crates/taskito-node/src/dispatcher.rs
+++ b/crates/taskito-node/src/dispatcher.rs
@@ -5,6 +5,7 @@
 //! `Promise<Buffer>`, and reports a [`JobResult`] back to the scheduler. This is
 //! the Node mirror of the Python shell's worker pool.
 
+use std::sync::Arc;
 use std::time::{Duration, Instant};
 
 use crossbeam_channel::Sender;
@@ -14,7 +15,7 @@ use taskito_core::job::Job;
 use taskito_core::scheduler::JobResult;
 use taskito_core::worker::WorkerDispatcher;
 use taskito_core::{Storage, StorageBackend};
-use tokio::sync::oneshot;
+use tokio::sync::{oneshot, Semaphore};
 
 use crate::convert::JsTaskInvocation;
 
@@ -25,11 +26,31 @@ type TaskCallback = ThreadsafeFunction<JsTaskInvocation, ErrorStrategy::Fatal>;
 pub struct NodeDispatcher {
     callback: TaskCallback,
     storage: StorageBackend,
+    /// Caps jobs running at once. Without it the loop spawns every job it is
+    /// handed and immediately takes the next, so nothing bounds concurrency.
+    ///
+    /// The scheduler's `max_in_flight` bounds what this worker *claims*; this
+    /// bounds what it *runs*, and is the only bound on the mesh path, where
+    /// stolen jobs never pass through this scheduler's in-flight accounting.
+    concurrency: Arc<Semaphore>,
 }
 
 impl NodeDispatcher {
-    pub fn new(callback: TaskCallback, storage: StorageBackend) -> Self {
-        Self { callback, storage }
+    /// `concurrency` of `None` leaves execution unbounded, matching the
+    /// behaviour of a worker that never set the option.
+    pub fn new(
+        callback: TaskCallback,
+        storage: StorageBackend,
+        concurrency: Option<usize>,
+    ) -> Self {
+        let permits = concurrency
+            .map(|c| c.max(1))
+            .unwrap_or(Semaphore::MAX_PERMITS);
+        Self {
+            callback,
+            storage,
+            concurrency: Arc::new(Semaphore::new(permits)),
+        }
     }
 }
 
@@ -41,13 +62,22 @@ impl WorkerDispatcher for NodeDispatcher {
         result_tx: Sender<JobResult>,
     ) {
         // Run jobs concurrently — each invocation is independent and the JS
-        // side may be async. The scheduler bounds in-flight work via the
-        // channel capacity and per-task/queue concurrency gates.
+        // side may be async — but take a permit first. Spawning unconditionally
+        // drains the job channel as fast as the scheduler fills it, so the
+        // channel applies no backpressure and in-flight work is unbounded: the
+        // worker claims jobs it cannot run, stranding them Running and starving
+        // peers on the same database. Blocking here is what bounds it; the
+        // permit rides with the job and is released when the task finishes.
         while let Some(job) = job_rx.recv().await {
+            let permit = match self.concurrency.clone().acquire_owned().await {
+                Ok(p) => p,
+                Err(_) => break,
+            };
             let callback = self.callback.clone();
             let storage = self.storage.clone();
             let result_tx = result_tx.clone();
             spawn(async move {
+                let _permit = permit;
                 let job_id = job.id.clone();
                 let result = run_one(&callback, &storage, job).await;
                 // A full bounded channel parks the sender — do it on the
diff --git a/crates/taskito-node/src/worker.rs b/crates/taskito-node/src/worker.rs
index ec672c15..90365a88 100644
--- a/crates/taskito-node/src/worker.rs
+++ b/crates/taskito-node/src/worker.rs
@@ -86,10 +86,20 @@ pub fn start_worker(
         .map(|c| (c as usize).max(1))
         .unwrap_or(DEFAULT_CHANNEL_CAPACITY);
 
+    // Jobs this worker runs at once. Separate from `channel_capacity`, which
+    // buffers hand-offs and advertises the worker's capacity — matching Java,
+    // where the two are distinct options.
+    let concurrency = options.concurrency.map(|c| (c as usize).max(1));
+
     let mut config = SchedulerConfig::default();
     if let Some(batch) = options.batch_size {
         config.batch_size = batch.max(1) as usize;
     }
+    // Bound in-flight work to what this worker will actually run, so it never
+    // claims more and strands the surplus Running while peers sharing the
+    // database skip it. Left unbounded when unset, so an existing worker does
+    // not silently acquire a cap it never asked for.
+    config.max_in_flight = concurrency;
 
     // The dispatcher reads cancel flags, and the lifecycle loop registers/heartbeats
     // — both need their own storage handle before `storage` moves into the scheduler.
@@ -178,7 +188,7 @@ pub fn start_worker(
     }
 
     // Dispatcher loop: execute each job in JS, report results on `result_tx`.
-    let dispatcher = NodeDispatcher::new(callback, dispatcher_storage);
+    let dispatcher = NodeDispatcher::new(callback, dispatcher_storage, concurrency);
     spawn(async move {
         dispatcher.run(job_rx, result_tx).await;
     });
diff --git a/sdks/node/src/types.ts b/sdks/node/src/types.ts
index eb3e2cdc..0c7573bf 100644
--- a/sdks/node/src/types.ts
+++ b/sdks/node/src/types.ts
@@ -160,8 +160,18 @@ export interface RegisteredTask {
 export interface WorkerRunOptions {
   /** Queues to consume (default `["default"]`). */
   queues?: string[];
-  /** In-flight channel capacity (default 128). */
+  /**
+   * Buffer size for job hand-offs, and the capacity this worker advertises
+   * (default 128). This does **not** bound how many jobs run at once — see
+   * {@link WorkerRunOptions.concurrency}.
+   */
   channelCapacity?: number;
+  /**
+   * Jobs this worker runs at once. Unset leaves it unbounded, so the worker can
+   * claim more than it can run, stranding the surplus `running` while peers
+   * sharing the database skip it. Set this on any worker sharing a database.
+   */
+  concurrency?: number;
   /** Jobs claimed per scheduler poll (default 1). */
   batchSize?: number;
   /**
diff --git a/sdks/node/src/worker.ts b/sdks/node/src/worker.ts
index 7b78c165..d5619709 100644
--- a/sdks/node/src/worker.ts
+++ b/sdks/node/src/worker.ts
@@ -225,6 +225,7 @@ export class Worker {
     const nativeOptions: NativeWorkerOptions = {
       queues: run?.queues,
       channelCapacity: run?.channelCapacity,
+      concurrency: run?.concurrency,
       batchSize: run?.batchSize,
       taskConfigs: applyTaskOverrides(
         buildTaskConfigs(tasks),
diff --git a/sdks/node/test/worker/backpressure.test.ts b/sdks/node/test/worker/backpressure.test.ts
new file mode 100644
index 00000000..f198e9dc
--- /dev/null
+++ b/sdks/node/test/worker/backpressure.test.ts
@@ -0,0 +1,65 @@
+import { mkdtempSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { afterEach, expect, it } from "vitest";
+import { Queue, type Worker } from "../../src/index";
+
+let worker: Worker | undefined;
+
+afterEach(() => {
+  worker?.stop();
+  worker = undefined;
+});
+
+function newQueue(): Queue {
+  return new Queue({ dbPath: join(mkdtempSync(join(tmpdir(), "taskito-bp-")), "q.db") });
+}
+
+async function waitFor(predicate: () => boolean, timeoutMs = 14000): Promise<boolean> {
+  const deadline = Date.now() + timeoutMs;
+  while (Date.now() < deadline) {
+    if (predicate()) {
+      return true;
+    }
+    await new Promise((resolve) => setTimeout(resolve, 20));
+  }
+  return false;
+}
+
+it("never runs more jobs at once than concurrency", async () => {
+  // The dispatcher used to spawn every job it was handed and immediately take
+  // the next, so the job channel drained as fast as the scheduler filled it and
+  // nothing bounded in-flight work — the worker claimed jobs it could not run,
+  // stranding them Running and starving peers on the same database.
+  //
+  // batchSize is the whole backlog on purpose. Several bounds overlap here —
+  // poll cadence (~50ms), task duration, channel capacity, the in-flight cap —
+  // and the test only proves something if the one under test is the tightest.
+  // Claiming all 12 in a single poll removes cadence from the picture: unfixed,
+  // all 12 dispatch at once and peak is 12; fixed, the in-flight clamp holds the
+  // claim to 3 regardless of how fast the poller runs.
+  const queue = newQueue();
+  const total = 12;
+  const concurrency = 3;
+  let running = 0;
+  let peak = 0;
+  let done = 0;
+
+  queue.task("slow", async () => {
+    running += 1;
+    peak = Math.max(peak, running);
+    await new Promise((resolve) => setTimeout(resolve, 400));
+    running -= 1;
+    done += 1;
+  });
+
+  for (let i = 0; i < total; i += 1) {
+    await queue.enqueue("slow");
+  }
+
+  worker = queue.runWorker({ queues: ["default"], concurrency, batchSize: total });
+
+  expect(await waitFor(() => done >= total)).toBe(true);
+  expect(peak).toBeLessThanOrEqual(concurrency);
+  expect(peak).toBeGreaterThan(1);
+});

From 326683c8e141c49a949edd3fe17ebb55446064c8 Mon Sep 17 00:00:00 2001
From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com>
Date: Thu, 16 Jul 2026 10:46:53 +0530
Subject: [PATCH 02/13] feat(node): surface maxInFlightPerTask
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Python has had this since the per-task bulkhead landed; this SDK hardcoded
None, so one slow task could occupy a worker's whole pool.

The config collector gated on a hand-listed set of option names, which drops
any option missing from the list — silently, and invisibly to type-checking.
Derive the gate from the built config instead, so a new option cannot be lost
the next time. Name the field in the rate-spec error too, now that more than
one option shares that grammar.
---
 crates/taskito-node/src/config.rs             |  3 +
 .../taskito-node/src/convert/task_config.rs   | 16 ++---
 sdks/node/src/types.ts                        |  9 ++-
 sdks/node/src/worker.ts                       | 50 ++++++++------
 sdks/node/test/worker/taskConfig.test.ts      | 67 +++++++++++++++++++
 5 files changed, 117 insertions(+), 28 deletions(-)
 create mode 100644 sdks/node/test/worker/taskConfig.test.ts

diff --git a/crates/taskito-node/src/config.rs b/crates/taskito-node/src/config.rs
index 9c2a8b1d..fe27a7eb 100644
--- a/crates/taskito-node/src/config.rs
+++ b/crates/taskito-node/src/config.rs
@@ -109,6 +109,9 @@ pub struct TaskConfigInput {
     pub retry_base_delay_ms: Option<i64>,
     pub retry_max_delay_ms: Option<i64>,
     pub max_concurrent: Option<i32>,
+    /// Cap on this task's share of one worker's dispatch slots, so a slow task
+    /// cannot occupy the whole pool. In-process, unlike `max_concurrent`.
+    pub max_in_flight_per_task: Option<i32>,
     /// Rate-limit spec like `"100/m"`, `"50/s"`, `"3600/h"`.
     pub rate_limit: Option<String>,
     pub circuit_breaker: Option<CircuitBreakerInput>,
diff --git a/crates/taskito-node/src/convert/task_config.rs b/crates/taskito-node/src/convert/task_config.rs
index 9bc63837..5784ea8c 100644
--- a/crates/taskito-node/src/convert/task_config.rs
+++ b/crates/taskito-node/src/convert/task_config.rs
@@ -15,13 +15,14 @@ const DEFAULT_MAX_RETRIES: i32 = 3;
 const DEFAULT_HALF_OPEN_PROBES: i32 = 5;
 const DEFAULT_HALF_OPEN_SUCCESS_RATE: f64 = 0.8;
 
-/// Parse an optional rate-limit spec, failing fast on a malformed value rather
-/// than silently disabling throttling (a misconfigured limit is a config error).
-fn parse_rate_limit(spec: Option<&str>) -> Result<Option<RateLimitConfig>> {
+/// Parse an optional rate spec, failing fast on a malformed value rather than
+/// silently disabling the cap (a misconfigured limit is a config error).
+/// `field` names the option in the error, since several share this grammar.
+fn parse_rate_spec(field: &str, spec: Option<&str>) -> Result<Option<RateLimitConfig>> {
     match spec {
         Some(s) => RateLimitConfig::parse(s)
             .map(Some)
-            .ok_or_else(|| invalid_arg(format!("invalid rateLimit '{s}' (expected e.g. '100/m')"))),
+            .ok_or_else(|| invalid_arg(format!("invalid {field} '{s}' (expected e.g. '100/m')"))),
         None => Ok(None),
     }
 }
@@ -35,7 +36,7 @@ pub fn task_config(input: &TaskConfigInput) -> Result<TaskConfig> {
             max_delay_ms: input.retry_max_delay_ms.unwrap_or(DEFAULT_RETRY_MAX_MS),
             custom_delays_ms: None,
         },
-        rate_limit: parse_rate_limit(input.rate_limit.as_deref())?,
+        rate_limit: parse_rate_spec("rateLimit", input.rate_limit.as_deref())?,
         circuit_breaker: input
             .circuit_breaker
             .as_ref()
@@ -44,8 +45,7 @@ pub fn task_config(input: &TaskConfigInput) -> Result<TaskConfig> {
         // 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,
+        max_in_flight_per_task: input.max_in_flight_per_task.map(|n| n.max(1) as usize),
     })
 }
 
@@ -88,7 +88,7 @@ fn circuit_breaker_config(input: &CircuitBreakerInput) -> Result<CircuitBreakerC
 /// Build a [`QueueConfig`] (rate limit, concurrency cap) from JS input.
 pub fn queue_config(input: &QueueConfigInput) -> Result<QueueConfig> {
     Ok(QueueConfig {
-        rate_limit: parse_rate_limit(input.rate_limit.as_deref())?,
+        rate_limit: parse_rate_spec("rateLimit", input.rate_limit.as_deref())?,
         max_concurrent: input.max_concurrent,
     })
 }
diff --git a/sdks/node/src/types.ts b/sdks/node/src/types.ts
index 0c7573bf..9dbf07b7 100644
--- a/sdks/node/src/types.ts
+++ b/sdks/node/src/types.ts
@@ -98,8 +98,15 @@ export interface TaskOptions {
   retryBackoff?: { baseMs?: number; maxMs?: number };
   /** Per-job timeout default (ms); enforced by the worker. */
   timeoutMs?: number;
-  /** Cap on concurrently-running jobs of this task. */
+  /** Cap on concurrently-running jobs of this task, across the cluster. */
   maxConcurrent?: number;
+  /**
+   * Cap on this task's share of a single worker's dispatch slots, so one slow
+   * task cannot occupy the whole pool and starve the others. In-process and
+   * free, unlike {@link TaskOptions.maxConcurrent}, which is cluster-wide and
+   * costs a database read.
+   */
+  maxInFlightPerTask?: number;
   /** Rate-limit spec like `"100/m"`, `"50/s"`, `"3600/h"`. */
   rateLimit?: RateLimit;
   /** Trip the task's circuit breaker after repeated failures. */
diff --git a/sdks/node/src/worker.ts b/sdks/node/src/worker.ts
index d5619709..330b50ca 100644
--- a/sdks/node/src/worker.ts
+++ b/sdks/node/src/worker.ts
@@ -21,7 +21,7 @@ import type {
 import { type ResourceRuntime, runWithResolver } from "./resources";
 import { deserializeCall, type PayloadCodec, type Serializer } from "./serializers";
 import { encodeTaskError } from "./task-error";
-import type { QueueLimits, RegisteredTask, WorkerRunOptions } from "./types";
+import type { QueueLimits, RegisteredTask, TaskOptions, WorkerRunOptions } from "./types";
 import { createLogger } from "./utils";
 import type { WorkflowTracker } from "./workflows";
 import { CACHE_TASK } from "./workflows/cache";
@@ -313,30 +313,42 @@ export class Worker {
 function buildTaskConfigs(tasks: ReadonlyMap<string, RegisteredTask>): TaskConfigInput[] {
   const configs: TaskConfigInput[] = [];
   for (const [name, task] of tasks) {
-    const options = task.options;
-    if (
-      !options ||
-      (options.maxRetries === undefined &&
-        options.retryBackoff === undefined &&
-        options.maxConcurrent === undefined &&
-        options.rateLimit === undefined &&
-        options.circuitBreaker === undefined)
-    ) {
+    if (!task.options) {
       continue;
     }
-    configs.push({
-      name,
-      maxRetries: options.maxRetries,
-      retryBaseDelayMs: options.retryBackoff?.baseMs,
-      retryMaxDelayMs: options.retryBackoff?.maxMs,
-      maxConcurrent: options.maxConcurrent,
-      rateLimit: options.rateLimit,
-      circuitBreaker: options.circuitBreaker,
-    });
+    const config = toTaskConfig(name, task.options);
+    if (setsSomething(config)) {
+      configs.push(config);
+    }
   }
   return configs;
 }
 
+function toTaskConfig(name: string, options: TaskOptions): TaskConfigInput {
+  return {
+    name,
+    maxRetries: options.maxRetries,
+    retryBaseDelayMs: options.retryBackoff?.baseMs,
+    retryMaxDelayMs: options.retryBackoff?.maxMs,
+    maxConcurrent: options.maxConcurrent,
+    maxInFlightPerTask: options.maxInFlightPerTask,
+    rateLimit: options.rateLimit,
+    circuitBreaker: options.circuitBreaker,
+  };
+}
+
+/**
+ * Whether a task set any policy worth registering.
+ *
+ * Derived from the built config rather than a hand-listed set of option names:
+ * a list silently drops any option missing from it, so a task setting only the
+ * new option would never reach the scheduler — with no error, and invisible to
+ * type-checking. `name` is always present, so it can't stand in for a setting.
+ */
+function setsSomething({ name: _name, ...policy }: TaskConfigInput): boolean {
+  return Object.values(policy).some((value) => value !== undefined);
+}
+
 function buildQueueConfigs(limits: ReadonlyMap<string, QueueLimits>): QueueConfigInput[] {
   return [...limits].map(([name, limit]) => ({
     name,
diff --git a/sdks/node/test/worker/taskConfig.test.ts b/sdks/node/test/worker/taskConfig.test.ts
new file mode 100644
index 00000000..54bdfb8c
--- /dev/null
+++ b/sdks/node/test/worker/taskConfig.test.ts
@@ -0,0 +1,67 @@
+import { mkdtempSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { afterEach, expect, it } from "vitest";
+import { Queue, type Worker } from "../../src/index";
+
+let worker: Worker | undefined;
+
+afterEach(() => {
+  worker?.stop();
+  worker = undefined;
+});
+
+function newQueue(): Queue {
+  return new Queue({ dbPath: join(mkdtempSync(join(tmpdir(), "taskito-tc-")), "q.db") });
+}
+
+async function waitFor(predicate: () => boolean, timeoutMs = 10000): Promise<boolean> {
+  const deadline = Date.now() + timeoutMs;
+  while (Date.now() < deadline) {
+    if (predicate()) {
+      return true;
+    }
+    await new Promise((resolve) => setTimeout(resolve, 20));
+  }
+  return false;
+}
+
+it("registers a task that sets only maxInFlightPerTask", async () => {
+  // The config collector used to gate on a hand-listed set of option names, so
+  // a task setting only an option missing from that list was dropped before it
+  // reached the scheduler — silently, and invisibly to type-checking. A job
+  // still runs either way, so assert the option reaches native by giving it a
+  // value the scheduler must honour: one slot means strictly serial execution.
+  const queue = newQueue();
+  const total = 4;
+  let running = 0;
+  let peak = 0;
+  let done = 0;
+
+  queue.task(
+    "solo",
+    async () => {
+      running += 1;
+      peak = Math.max(peak, running);
+      await new Promise((resolve) => setTimeout(resolve, 150));
+      running -= 1;
+      done += 1;
+    },
+    { maxInFlightPerTask: 1 },
+  );
+
+  for (let i = 0; i < total; i += 1) {
+    await queue.enqueue("solo");
+  }
+
+  worker = queue.runWorker({ queues: ["default"], concurrency: 4, batchSize: total });
+
+  expect(await waitFor(() => done >= total)).toBe(true);
+  expect(peak).toBe(1);
+});
+
+it("rejects a malformed rateLimit rather than silently disabling it", async () => {
+  const queue = newQueue();
+  queue.task("bad", async () => {}, { rateLimit: "100/mm" as never });
+  expect(() => queue.runWorker({ queues: ["default"] })).toThrow(/rateLimit/);
+});

From a682e4b9f1be8a8d1083486d7270427966759e84 Mon Sep 17 00:00:00 2001
From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com>
Date: Thu, 16 Jul 2026 11:01:54 +0530
Subject: [PATCH 03/13] feat(java): surface per-task rate limit and concurrency
 cap
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Python and Node have had both for a while; this SDK carried neither. The
hardcoded Nones were not a lazy binding — the wire struct had no field for
them, so nothing upstream could send one.

Available on Task and on @TaskHandler, matching how the retry curve and
circuit breaker already reach the worker. A malformed rate spec now fails the
worker's start, naming the task and option, rather than running the task
unthrottled; registration gained an error path to allow that.

Worker holds the tasks that carry policy rather than a map per knob, so the
encoder reads one source: the next knob is encoded in one place instead of
needing its own map, its own capture, and its own arm of a name union.
---
 crates/taskito-java/src/convert.rs            |  10 +-
 crates/taskito-java/src/worker.rs             |  44 +++++--
 .../processor/TaskHandlerProcessor.java       |   8 ++
 .../taskito/annotation/TaskHandler.java       |  13 ++
 .../java/org/byteveda/taskito/task/Task.java  |  65 ++++++++--
 .../org/byteveda/taskito/worker/Worker.java   |  96 +++++++++------
 .../taskito/core/TaskPolicyConfigTest.java    | 112 ++++++++++++++++++
 7 files changed, 290 insertions(+), 58 deletions(-)
 create mode 100644 sdks/java/src/test/java/org/byteveda/taskito/core/TaskPolicyConfigTest.java

diff --git a/crates/taskito-java/src/convert.rs b/crates/taskito-java/src/convert.rs
index 6d6e4888..a520ba2b 100644
--- a/crates/taskito-java/src/convert.rs
+++ b/crates/taskito-java/src/convert.rs
@@ -319,9 +319,9 @@ pub struct SubscriptionSpec {
     pub timeout_ms: Option<i64>,
 }
 
-/// A task's retry-backoff curve. Fields left unset fall back to the core's
-/// `RetryPolicy` defaults; the per-job retry budget travels via `maxRetries` on
-/// enqueue, so it is deliberately absent here.
+/// A task's policy: retry curve, throttling, and concurrency caps. Fields left
+/// unset fall back to the core's defaults. The per-job retry *budget* travels
+/// via `maxRetries` on enqueue, so it is deliberately absent here.
 #[derive(Deserialize)]
 #[serde(rename_all = "camelCase")]
 pub struct TaskRetryConfig {
@@ -336,6 +336,10 @@ pub struct TaskRetryConfig {
     pub circuit_breaker_cooldown_ms: Option<i64>,
     pub circuit_breaker_half_open_probes: Option<i32>,
     pub circuit_breaker_half_open_success_rate: Option<f64>,
+    /// Rate-limit spec like `"100/m"`, `"50/s"`, `"3600/h"`.
+    pub rate_limit: Option<String>,
+    /// Cap on concurrently-running jobs of this task, across the cluster.
+    pub max_concurrent: Option<i32>,
 }
 
 /// Filter accepted by `NativeQueue.listJobs`.
diff --git a/crates/taskito-java/src/worker.rs b/crates/taskito-java/src/worker.rs
index d5f8d69b..db93254e 100644
--- a/crates/taskito-java/src/worker.rs
+++ b/crates/taskito-java/src/worker.rs
@@ -13,6 +13,7 @@ use jni::objects::{GlobalRef, JByteArray, JClass, JObject, JString, JValue};
 use jni::sys::jlong;
 use jni::JNIEnv;
 use taskito_core::resilience::circuit_breaker::CircuitBreakerConfig;
+use taskito_core::resilience::rate_limiter::RateLimitConfig;
 use taskito_core::resilience::retry::RetryPolicy;
 use taskito_core::scheduler::{ResultOutcome, TaskConfig};
 use taskito_core::worker::WorkerDispatcher;
@@ -105,7 +106,7 @@ fn start_worker(
     // Claim execution under this worker's id so dead-worker recovery can
     // attribute orphaned jobs (matches the worker id registered above).
     scheduler.set_claim_owner(worker_id.clone());
-    register_task_policies(&mut scheduler, options.task_configs);
+    register_task_policies(&mut scheduler, options.task_configs)?;
     let scheduler = Arc::new(scheduler);
     let shutdown = scheduler.shutdown_handle();
     let heartbeat_stop = Arc::new(Notify::new());
@@ -234,11 +235,19 @@ fn register_subscriptions(
     Ok(())
 }
 
-/// Register each task's retry-backoff curve with the scheduler. Only the curve
-/// is set here — the core resolves the retry budget from the job's `max_retries`
-/// — so unset fields keep the core [`RetryPolicy`] defaults.
-fn register_task_policies(scheduler: &mut Scheduler, configs: Option<Vec<TaskRetryConfig>>) {
-    let Some(configs) = configs else { return };
+/// Register each task's policy — retry curve, throttling, concurrency caps —
+/// with the scheduler. Unset fields keep the core's defaults; the per-job retry
+/// budget still resolves from the job's `max_retries`.
+///
+/// Fails on a malformed rate spec rather than registering the task without it:
+/// silently dropping a throttle the caller asked for is worse than not starting.
+fn register_task_policies(
+    scheduler: &mut Scheduler,
+    configs: Option<Vec<TaskRetryConfig>>,
+) -> Result<(), crate::error::BindingError> {
+    let Some(configs) = configs else {
+        return Ok(());
+    };
     for config in configs {
         let mut retry_policy = RetryPolicy::default();
         if let Some(custom) = config.custom_delays_ms {
@@ -271,20 +280,39 @@ fn register_task_policies(scheduler: &mut Scheduler, configs: Option<Vec<TaskRet
                         .circuit_breaker_half_open_success_rate
                         .unwrap_or(0.8),
                 });
+        let rate_limit = parse_rate_spec("rateLimit", &config.name, config.rate_limit.as_deref())?;
         scheduler.register_task(
             config.name,
             TaskConfig {
                 retry_policy,
-                rate_limit: None,
+                rate_limit,
                 circuit_breaker,
                 // Not surfaced on this SDK yet; retries stay bounded per job.
                 retry_budget: None,
-                max_concurrent: None,
+                max_concurrent: config.max_concurrent,
                 // Not surfaced on this SDK yet; the whole pool stays available.
                 max_in_flight_per_task: None,
             },
         );
     }
+    Ok(())
+}
+
+/// Parse an optional rate spec, naming the offending task and option so a typo
+/// is actionable. Several options share this `"100/m"` grammar.
+fn parse_rate_spec(
+    field: &str,
+    task: &str,
+    spec: Option<&str>,
+) -> Result<Option<RateLimitConfig>, crate::error::BindingError> {
+    match spec {
+        Some(s) => RateLimitConfig::parse(s).map(Some).ok_or_else(|| {
+            crate::error::BindingError::new(format!(
+                "invalid {field} '{s}' on task '{task}' (expected e.g. '100/m')"
+            ))
+        }),
+        None => Ok(None),
+    }
 }
 
 /// Drain results: apply each to storage and invoke `WorkerBridge.onOutcome`.
diff --git a/sdks/java/processor/src/main/java/org/byteveda/taskito/processor/TaskHandlerProcessor.java b/sdks/java/processor/src/main/java/org/byteveda/taskito/processor/TaskHandlerProcessor.java
index dad14bda..e82bdf54 100644
--- a/sdks/java/processor/src/main/java/org/byteveda/taskito/processor/TaskHandlerProcessor.java
+++ b/sdks/java/processor/src/main/java/org/byteveda/taskito/processor/TaskHandlerProcessor.java
@@ -198,6 +198,14 @@ private String options(ExecutableElement method) {
         if (boolValue(mirror, "idempotent", false)) {
             chain.append(".idempotent(true)");
         }
+        String rateLimit = stringValue(mirror, "rateLimit", "");
+        if (!rateLimit.isEmpty()) {
+            chain.append(".rateLimit(\"").append(rateLimit).append("\")");
+        }
+        long maxConcurrent = longValue(mirror, "maxConcurrent", 0);
+        if (maxConcurrent > 0) {
+            chain.append(".maxConcurrent(").append(maxConcurrent).append(")");
+        }
         appendCircuitBreaker(chain, mirror);
         return chain.toString();
     }
diff --git a/sdks/java/src/main/java/org/byteveda/taskito/annotation/TaskHandler.java b/sdks/java/src/main/java/org/byteveda/taskito/annotation/TaskHandler.java
index 0dafa40d..6fe3a170 100644
--- a/sdks/java/src/main/java/org/byteveda/taskito/annotation/TaskHandler.java
+++ b/sdks/java/src/main/java/org/byteveda/taskito/annotation/TaskHandler.java
@@ -36,6 +36,19 @@
     /** Auto-derive an idempotency {@code uniqueKey} from the payload on every enqueue. */
     boolean idempotent() default false;
 
+    /**
+     * Rate-limit spec like {@code "100/m"} ({@code s}, {@code m} and {@code h}
+     * suffixes); empty (default) leaves the task unthrottled. A malformed spec
+     * fails the worker's start rather than running unthrottled.
+     */
+    String rateLimit() default "";
+
+    /**
+     * Cap on concurrently-running jobs of this task across the cluster; 0
+     * (default) leaves it uncapped.
+     */
+    int maxConcurrent() default 0;
+
     /** Circuit-breaker failure threshold; 0 (default) leaves the breaker off. */
     int circuitBreakerThreshold() default 0;
 
diff --git a/sdks/java/src/main/java/org/byteveda/taskito/task/Task.java b/sdks/java/src/main/java/org/byteveda/taskito/task/Task.java
index 34ac14a3..4460492b 100644
--- a/sdks/java/src/main/java/org/byteveda/taskito/task/Task.java
+++ b/sdks/java/src/main/java/org/byteveda/taskito/task/Task.java
@@ -22,6 +22,8 @@ public final class Task<T> {
     private final List<String> codecs;
     private final boolean idempotent;
     private final CircuitBreakerConfig circuitBreaker;
+    private final String rateLimit;
+    private final Integer maxConcurrent;
 
     private Task(
             String name,
@@ -30,7 +32,9 @@ private Task(
             RetryPolicy retryPolicy,
             List<String> codecs,
             boolean idempotent,
-            CircuitBreakerConfig circuitBreaker) {
+            CircuitBreakerConfig circuitBreaker,
+            String rateLimit,
+            Integer maxConcurrent) {
         this.name = Objects.requireNonNull(name, "task name must not be null");
         if (name.trim().isEmpty()) {
             throw new IllegalArgumentException("task name must not be blank");
@@ -41,21 +45,24 @@ private Task(
         this.codecs = List.copyOf(codecs);
         this.idempotent = idempotent;
         this.circuitBreaker = circuitBreaker;
+        this.rateLimit = rateLimit;
+        this.maxConcurrent = maxConcurrent;
     }
 
     /** A task whose payload deserializes to {@code payloadType}. */
     public static <T> Task<T> of(String name, Class<T> payloadType) {
-        return new Task<>(name, payloadType, EnqueueOptions.none(), null, List.of(), false, null);
+        return new Task<>(name, payloadType, EnqueueOptions.none(), null, List.of(), false, null, null, null);
     }
 
     /** A task whose payload deserializes to a generic type, e.g. {@code new TypeReference<List<Foo>>(){}}. */
     public static <T> Task<T> of(String name, TypeReference<T> payloadType) {
-        return new Task<>(name, payloadType.getType(), EnqueueOptions.none(), null, List.of(), false, null);
+        return new Task<>(name, payloadType.getType(), EnqueueOptions.none(), null, List.of(), false, null, null, null);
     }
 
     /** A copy of this task with the given default options. */
     public Task<T> withOptions(EnqueueOptions options) {
-        return new Task<>(name, payloadType, options, retryPolicy, codecs, idempotent, circuitBreaker);
+        return new Task<>(
+                name, payloadType, options, retryPolicy, codecs, idempotent, circuitBreaker, rateLimit, maxConcurrent);
     }
 
     /**
@@ -64,7 +71,8 @@ public Task<T> withOptions(EnqueueOptions options) {
      * from {@link #maxRetries}.
      */
     public Task<T> retryPolicy(RetryPolicy retryPolicy) {
-        return new Task<>(name, payloadType, options, retryPolicy, codecs, idempotent, circuitBreaker);
+        return new Task<>(
+                name, payloadType, options, retryPolicy, codecs, idempotent, circuitBreaker, rateLimit, maxConcurrent);
     }
 
     /**
@@ -74,7 +82,16 @@ public Task<T> retryPolicy(RetryPolicy retryPolicy) {
      * {@code Taskito.builder().codec(name, codec)} on producers and workers.
      */
     public Task<T> codecs(String... codecs) {
-        return new Task<>(name, payloadType, options, retryPolicy, Arrays.asList(codecs), idempotent, circuitBreaker);
+        return new Task<>(
+                name,
+                payloadType,
+                options,
+                retryPolicy,
+                Arrays.asList(codecs),
+                idempotent,
+                circuitBreaker,
+                rateLimit,
+                maxConcurrent);
     }
 
     /**
@@ -83,7 +100,8 @@ public Task<T> codecs(String... codecs) {
      * A per-enqueue {@link EnqueueOptions.Builder#idempotent(boolean)} overrides this default.
      */
     public Task<T> idempotent(boolean idempotent) {
-        return new Task<>(name, payloadType, options, retryPolicy, codecs, idempotent, circuitBreaker);
+        return new Task<>(
+                name, payloadType, options, retryPolicy, codecs, idempotent, circuitBreaker, rateLimit, maxConcurrent);
     }
 
     /**
@@ -92,7 +110,28 @@ public Task<T> idempotent(boolean idempotent) {
      * it recovers.
      */
     public Task<T> circuitBreaker(CircuitBreakerConfig circuitBreaker) {
-        return new Task<>(name, payloadType, options, retryPolicy, codecs, idempotent, circuitBreaker);
+        return new Task<>(
+                name, payloadType, options, retryPolicy, codecs, idempotent, circuitBreaker, rateLimit, maxConcurrent);
+    }
+
+    /**
+     * A copy of this task throttled to {@code rateLimit}, a spec like {@code "100/m"}
+     * ({@code s}, {@code m} and {@code h} suffixes). The worker registers it on
+     * {@code start()} and rejects a malformed spec rather than running unthrottled.
+     */
+    public Task<T> rateLimit(String rateLimit) {
+        return new Task<>(
+                name, payloadType, options, retryPolicy, codecs, idempotent, circuitBreaker, rateLimit, maxConcurrent);
+    }
+
+    /**
+     * A copy of this task allowed at most {@code maxConcurrent} jobs running at once
+     * across the cluster. The scheduler counts running jobs before dispatch, so this
+     * costs a database read; {@code null} means no cap.
+     */
+    public Task<T> maxConcurrent(Integer maxConcurrent) {
+        return new Task<>(
+                name, payloadType, options, retryPolicy, codecs, idempotent, circuitBreaker, rateLimit, maxConcurrent);
     }
 
     public Task<T> queue(String queue) {
@@ -160,4 +199,14 @@ public boolean idempotent() {
     public CircuitBreakerConfig circuitBreaker() {
         return circuitBreaker;
     }
+
+    /** This task's rate-limit spec (e.g. {@code "100/m"}), or {@code null} when unthrottled. */
+    public String rateLimit() {
+        return rateLimit;
+    }
+
+    /** Cap on this task's concurrently-running jobs, or {@code null} when uncapped. */
+    public Integer maxConcurrent() {
+        return maxConcurrent;
+    }
 }
diff --git a/sdks/java/src/main/java/org/byteveda/taskito/worker/Worker.java b/sdks/java/src/main/java/org/byteveda/taskito/worker/Worker.java
index 26da11b9..1a060e65 100644
--- a/sdks/java/src/main/java/org/byteveda/taskito/worker/Worker.java
+++ b/sdks/java/src/main/java/org/byteveda/taskito/worker/Worker.java
@@ -8,11 +8,9 @@
 import java.util.EnumMap;
 import java.util.HashMap;
 import java.util.LinkedHashMap;
-import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
-import java.util.Set;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
@@ -182,8 +180,14 @@ public static final class Builder {
         private final ResourceRuntime resources;
         private final Map<String, PayloadCodec> codecs;
         private final Map<String, RegisteredTask> handlers = new HashMap<>();
-        private final Map<String, RetryPolicy> taskPolicies = new HashMap<>();
-        private final Map<String, CircuitBreakerConfig> taskCircuitBreakers = new HashMap<>();
+        /**
+         * Tasks carrying policy the worker must register, keyed by name. Holding the
+         * task itself rather than a map per knob keeps {@link #encodeTaskConfigs}
+         * reading from one source: a new knob is then encoded in one place instead of
+         * needing its own map, its own capture, and its own arm of a name union.
+         */
+        private final Map<String, Task<?>> taskPolicies = new LinkedHashMap<>();
+
         private final Map<EventName, List<Consumer<OutcomeEvent>>> listeners = new EnumMap<>(EventName.class);
         private List<SubscriptionConfig> subscriptions = List.of();
         private List<String> queues;
@@ -236,18 +240,21 @@ public Builder register(Handler<?, ?> handler) {
             return this;
         }
 
-        /** Remember a task's retry-backoff curve and circuit breaker so {@code start()} registers them. */
+        /** Remember a task's policy — retries, breaker, throttling, caps — so {@code start()} registers it. */
         private void capturePolicy(Task<?> task) {
-            RetryPolicy policy = task.retryPolicy();
-            if (policy != null) {
-                taskPolicies.put(task.name(), policy);
-            }
-            CircuitBreakerConfig breaker = task.circuitBreaker();
-            if (breaker != null) {
-                taskCircuitBreakers.put(task.name(), breaker);
+            if (hasPolicy(task)) {
+                taskPolicies.put(task.name(), task);
             }
         }
 
+        /** Whether a task sets anything the scheduler needs registering. */
+        private static boolean hasPolicy(Task<?> task) {
+            return task.retryPolicy() != null
+                    || task.circuitBreaker() != null
+                    || task.rateLimit() != null
+                    || task.maxConcurrent() != null;
+        }
+
         /** Register every handler in a {@link HandlerRegistry} (e.g. a generated {@code XxxTasks.handlers}). */
         public Builder register(HandlerRegistry registry) {
             registry.handlers().forEach(this::register);
@@ -432,7 +439,7 @@ private String encodeOptions() {
             } else if (concurrency > 0) {
                 options.put("concurrency", concurrency);
             }
-            if (!taskPolicies.isEmpty() || !taskCircuitBreakers.isEmpty()) {
+            if (!taskPolicies.isEmpty()) {
                 options.put("taskConfigs", encodeTaskConfigs());
             }
             if (mesh != null) {
@@ -481,40 +488,51 @@ private List<Map<String, Object>> encodeSubscriptions() {
          * wire shape the binding reads — one entry per task, merging both sources by name.
          */
         private List<Map<String, Object>> encodeTaskConfigs() {
-            Set<String> names = new LinkedHashSet<>(taskPolicies.keySet());
-            names.addAll(taskCircuitBreakers.keySet());
-            List<Map<String, Object>> configs = new ArrayList<>(names.size());
-            for (String name : names) {
+            List<Map<String, Object>> configs = new ArrayList<>(taskPolicies.size());
+            for (Task<?> task : taskPolicies.values()) {
                 Map<String, Object> config = new LinkedHashMap<>();
-                config.put("name", name);
-                RetryPolicy policy = taskPolicies.get(name);
-                if (policy != null) {
-                    if (policy.baseDelay() != null) {
-                        config.put("baseDelayMs", policy.baseDelay().toMillis());
-                    }
-                    if (policy.maxDelay() != null) {
-                        config.put("maxDelayMs", policy.maxDelay().toMillis());
-                    }
-                    if (!policy.customDelays().isEmpty()) {
-                        List<Long> delaysMs =
-                                new ArrayList<>(policy.customDelays().size());
-                        policy.customDelays().forEach(delay -> delaysMs.add(delay.toMillis()));
-                        config.put("customDelaysMs", delaysMs);
-                    }
+                config.put("name", task.name());
+                encodeRetryPolicy(task.retryPolicy(), config);
+                encodeCircuitBreaker(task.circuitBreaker(), config);
+                if (task.rateLimit() != null) {
+                    config.put("rateLimit", task.rateLimit());
                 }
-                CircuitBreakerConfig breaker = taskCircuitBreakers.get(name);
-                if (breaker != null) {
-                    config.put("circuitBreakerThreshold", breaker.threshold());
-                    config.put("circuitBreakerWindowMs", breaker.window().toMillis());
-                    config.put("circuitBreakerCooldownMs", breaker.cooldown().toMillis());
-                    config.put("circuitBreakerHalfOpenProbes", breaker.halfOpenProbes());
-                    config.put("circuitBreakerHalfOpenSuccessRate", breaker.halfOpenSuccessRate());
+                if (task.maxConcurrent() != null) {
+                    config.put("maxConcurrent", task.maxConcurrent());
                 }
                 configs.add(config);
             }
             return configs;
         }
 
+        private static void encodeRetryPolicy(RetryPolicy policy, Map<String, Object> config) {
+            if (policy == null) {
+                return;
+            }
+            if (policy.baseDelay() != null) {
+                config.put("baseDelayMs", policy.baseDelay().toMillis());
+            }
+            if (policy.maxDelay() != null) {
+                config.put("maxDelayMs", policy.maxDelay().toMillis());
+            }
+            if (!policy.customDelays().isEmpty()) {
+                List<Long> delaysMs = new ArrayList<>(policy.customDelays().size());
+                policy.customDelays().forEach(delay -> delaysMs.add(delay.toMillis()));
+                config.put("customDelaysMs", delaysMs);
+            }
+        }
+
+        private static void encodeCircuitBreaker(CircuitBreakerConfig breaker, Map<String, Object> config) {
+            if (breaker == null) {
+                return;
+            }
+            config.put("circuitBreakerThreshold", breaker.threshold());
+            config.put("circuitBreakerWindowMs", breaker.window().toMillis());
+            config.put("circuitBreakerCooldownMs", breaker.cooldown().toMillis());
+            config.put("circuitBreakerHalfOpenProbes", breaker.halfOpenProbes());
+            config.put("circuitBreakerHalfOpenSuccessRate", breaker.halfOpenSuccessRate());
+        }
+
         @SuppressWarnings("unchecked")
         private static <T, R> TaskFunction<Object, Object> cast(TaskFunction<T, R> handler) {
             return (TaskFunction<Object, Object>) handler;
diff --git a/sdks/java/src/test/java/org/byteveda/taskito/core/TaskPolicyConfigTest.java b/sdks/java/src/test/java/org/byteveda/taskito/core/TaskPolicyConfigTest.java
new file mode 100644
index 00000000..a6b4baf9
--- /dev/null
+++ b/sdks/java/src/test/java/org/byteveda/taskito/core/TaskPolicyConfigTest.java
@@ -0,0 +1,112 @@
+package org.byteveda.taskito.core;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.nio.file.Path;
+import java.time.Duration;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.byteveda.taskito.Taskito;
+import org.byteveda.taskito.events.EventName;
+import org.byteveda.taskito.task.Task;
+import org.byteveda.taskito.worker.Worker;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.api.io.TempDir;
+
+/** Per-task throttling and concurrency caps reach the scheduler. */
+class TaskPolicyConfigTest {
+
+    /** A cap of one means jobs of this task never overlap, however many the worker runs. */
+    @Test
+    @Timeout(30)
+    void maxConcurrentSerializesOneTask(@TempDir Path dir) throws Exception {
+        Task<String> solo = Task.of("solo", String.class).maxConcurrent(1);
+        AtomicInteger running = new AtomicInteger();
+        AtomicInteger peak = new AtomicInteger();
+        CountDownLatch done = new CountDownLatch(4);
+
+        try (Taskito queue = Taskito.builder()
+                .backend("sqlite")
+                .url(dir.resolve("t.db").toString())
+                .open()) {
+            for (int i = 0; i < 4; i++) {
+                queue.enqueue(solo, String.valueOf(i));
+            }
+
+            try (Worker worker = queue.worker()
+                    .concurrency(4)
+                    .handle(solo, (String p) -> {
+                        peak.accumulateAndGet(running.incrementAndGet(), Math::max);
+                        try {
+                            Thread.sleep(150);
+                        } catch (InterruptedException e) {
+                            Thread.currentThread().interrupt();
+                        }
+                        running.decrementAndGet();
+                        return null;
+                    })
+                    .on(EventName.SUCCESS, event -> done.countDown())
+                    .start()) {
+                assertTrue(done.await(20, TimeUnit.SECONDS), "all four jobs should finish");
+            }
+        }
+
+        assertEquals(1, peak.get(), "maxConcurrent=1 must never let two jobs of the task overlap");
+    }
+
+    /** Throttling drags a burst out past the window it would otherwise finish in. */
+    @Test
+    @Timeout(30)
+    void rateLimitThrottlesDispatch(@TempDir Path dir) throws Exception {
+        // The bucket starts full, so the spec's count is also its burst: "60/m"
+        // would let 60 jobs through at once and prove nothing. "1/s" holds one
+        // token, so of three jobs the first runs at once and the rest wait for a
+        // refill — roughly a second each.
+        Task<String> limited = Task.of("limited", String.class).rateLimit("1/s");
+        CountDownLatch done = new CountDownLatch(3);
+
+        try (Taskito queue = Taskito.builder()
+                .backend("sqlite")
+                .url(dir.resolve("t.db").toString())
+                .open()) {
+            for (int i = 0; i < 3; i++) {
+                queue.enqueue(limited, String.valueOf(i));
+            }
+
+            long start = System.nanoTime();
+            try (Worker worker = queue.worker()
+                    .handle(limited, (String p) -> null)
+                    .on(EventName.SUCCESS, event -> done.countDown())
+                    .start()) {
+                assertTrue(done.await(20, TimeUnit.SECONDS), "all three jobs should finish");
+            }
+            Duration elapsed = Duration.ofNanos(System.nanoTime() - start);
+            assertTrue(
+                    elapsed.toMillis() >= 1200,
+                    "a throttled burst must not finish instantly, took " + elapsed.toMillis() + "ms");
+        }
+    }
+
+    /** A typo must fail the start, not silently run the task unthrottled. */
+    @Test
+    @Timeout(30)
+    void malformedRateLimitFailsTheStart(@TempDir Path dir) {
+        Task<String> bad = Task.of("bad", String.class).rateLimit("100/mm");
+
+        try (Taskito queue = Taskito.builder()
+                .backend("sqlite")
+                .url(dir.resolve("t.db").toString())
+                .open()) {
+            RuntimeException error = assertThrows(
+                    RuntimeException.class,
+                    () -> queue.worker().handle(bad, (String p) -> null).start());
+            assertTrue(
+                    error.getMessage().contains("rateLimit"),
+                    "the error should name the offending option, got: " + error.getMessage());
+        }
+    }
+}

From c521c96f401edb6267c59505b3ecc4bc5859c8ba Mon Sep 17 00:00:00 2001
From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com>
Date: Thu, 16 Jul 2026 11:05:57 +0530
Subject: [PATCH 04/13] perf(node): finalize results in batches
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The drain loop handled one result per wake, so each job cost its own
transaction. Take everything already queued and finalize it in one, matching
Python. The channel is bounded, so the drain caps itself.

Successes batch into a single write while failures stay per-result, and one
outcome still comes back per job in order, so events and middleware fire
exactly once each. The catch_unwind now guards a batch, widening a panic from
one dropped outcome to N — acceptable because the realistic failure is a batch
write error, which already falls back to finalizing job by job.
---
 crates/taskito-node/src/worker.rs            | 44 +++++++++----
 sdks/node/test/worker/resultBatching.test.ts | 65 ++++++++++++++++++++
 2 files changed, 97 insertions(+), 12 deletions(-)
 create mode 100644 sdks/node/test/worker/resultBatching.test.ts

diff --git a/crates/taskito-node/src/worker.rs b/crates/taskito-node/src/worker.rs
index 90365a88..83126d1f 100644
--- a/crates/taskito-node/src/worker.rs
+++ b/crates/taskito-node/src/worker.rs
@@ -198,23 +198,43 @@ pub fn start_worker(
     // sender has dropped (i.e. after the dispatcher and all in-flight jobs end).
     let scheduler_results = scheduler;
     spawn_blocking(move || {
-        while let Ok(result) = result_rx.recv() {
-            // A panicking result must not kill the drain loop — a dead loop
-            // silently drops every later outcome.
+        while let Ok(first) = result_rx.recv() {
+            // Take everything already queued and finalize it in one batched
+            // transaction rather than one per wake. The channel is bounded, so
+            // the drain caps itself.
+            let mut batch = vec![first];
+            while let Ok(more) = result_rx.try_recv() {
+                batch.push(more);
+            }
+
+            // A panicking batch must not kill the drain loop — a dead loop
+            // silently drops every later outcome. The batch widens a panic's
+            // reach from one outcome to N, which is acceptable: the realistic
+            // failure is a batch write error, and `handle_results` already
+            // falls back to finalizing those jobs one at a time.
             let handled = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
-                scheduler_results.handle_result(result)
+                scheduler_results.handle_results(batch)
             }));
             match handled {
                 // Surface each outcome to JS so the shell can emit events and run
-                // middleware (the events layer).
-                Ok(Ok(outcome)) => {
-                    outcome_callback.call(
-                        outcome_to_js(&outcome),
-                        ThreadsafeFunctionCallMode::NonBlocking,
-                    );
+                // middleware (the events layer). `handle_results` returns one
+                // outcome per input, in order, so each job is reported once.
+                Ok(outcomes) => {
+                    for outcome in outcomes {
+                        match outcome {
+                            Ok(outcome) => {
+                                outcome_callback.call(
+                                    outcome_to_js(&outcome),
+                                    ThreadsafeFunctionCallMode::NonBlocking,
+                                );
+                            }
+                            Err(err) => {
+                                log::error!("[taskito-node] result handling error: {err}")
+                            }
+                        }
+                    }
                 }
-                Ok(Err(err)) => log::error!("[taskito-node] result handling error: {err}"),
-                Err(_) => log::error!("[taskito-node] result handling panicked; outcome dropped"),
+                Err(_) => log::error!("[taskito-node] result handling panicked; outcomes dropped"),
             }
         }
     });
diff --git a/sdks/node/test/worker/resultBatching.test.ts b/sdks/node/test/worker/resultBatching.test.ts
new file mode 100644
index 00000000..e5e9c6ba
--- /dev/null
+++ b/sdks/node/test/worker/resultBatching.test.ts
@@ -0,0 +1,65 @@
+import { mkdtempSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { afterEach, expect, it } from "vitest";
+import { Queue, type Worker } from "../../src/index";
+
+let worker: Worker | undefined;
+
+afterEach(() => {
+  worker?.stop();
+  worker = undefined;
+});
+
+function newQueue(): Queue {
+  return new Queue({ dbPath: join(mkdtempSync(join(tmpdir(), "taskito-rb-")), "q.db") });
+}
+
+async function waitFor(predicate: () => boolean, timeoutMs = 14000): Promise<boolean> {
+  const deadline = Date.now() + timeoutMs;
+  while (Date.now() < deadline) {
+    if (predicate()) {
+      return true;
+    }
+    await new Promise((resolve) => setTimeout(resolve, 20));
+  }
+  return false;
+}
+
+it("reports every job exactly once when results are drained as a batch", async () => {
+  // The drain loop finalizes everything already queued in one transaction, so a
+  // burst of results is handled as a batch rather than one per wake. Successes
+  // batch into a single write while failures stay on the per-result path, so a
+  // mixed burst exercises both halves: each job must still surface exactly one
+  // outcome, none lost and none double-counted.
+  const queue = newQueue();
+  const each = 6;
+  const completed: string[] = [];
+  const dead: string[] = [];
+
+  queue.on("job.completed", (event) => completed.push(event.jobId));
+  queue.on("job.dead", (event) => dead.push(event.jobId));
+
+  queue.task("ok", (n: number) => n);
+  queue.task(
+    "boom",
+    () => {
+      throw new Error("expected");
+    },
+    { maxRetries: 0 },
+  );
+
+  for (let i = 0; i < each; i += 1) {
+    queue.enqueue("ok", [i]);
+    queue.enqueue("boom");
+  }
+
+  worker = queue.runWorker({ queues: ["default"], concurrency: 8, batchSize: each * 2 });
+
+  expect(await waitFor(() => completed.length >= each && dead.length >= each)).toBe(true);
+
+  expect(completed).toHaveLength(each);
+  expect(dead).toHaveLength(each);
+  expect(new Set(completed).size).toBe(each);
+  expect(new Set(dead).size).toBe(each);
+});

From 0121d69c18ff31a8b22f1adb653391857c2d4d8e Mon Sep 17 00:00:00 2001
From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com>
Date: Thu, 16 Jul 2026 11:10:38 +0530
Subject: [PATCH 05/13] perf(java): finalize results in batches
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The drain loop handled one result per wake, so each job cost its own
transaction. Take everything already queued and finalize it in one, matching
Python and Node.

The batch is capped explicitly rather than drained until empty: unlike the
other shells, this result channel is unbounded so the dispatcher never blocks
a runtime worker, which means nothing else would bound one drain. The JNI
local frame stays per outcome — sized for a single callback, it would overflow
around a batch.
---
 crates/taskito-java/src/worker.rs             | 61 ++++++++++-----
 .../taskito/core/ResultBatchingTest.java      | 76 +++++++++++++++++++
 2 files changed, 118 insertions(+), 19 deletions(-)
 create mode 100644 sdks/java/src/test/java/org/byteveda/taskito/core/ResultBatchingTest.java

diff --git a/crates/taskito-java/src/worker.rs b/crates/taskito-java/src/worker.rs
index db93254e..7a06d044 100644
--- a/crates/taskito-java/src/worker.rs
+++ b/crates/taskito-java/src/worker.rs
@@ -157,7 +157,9 @@ fn start_worker(
     // result sender has dropped (dispatcher done).
     let drain_scheduler = scheduler;
     let drain_callbacks = callbacks;
-    runtime.spawn_blocking(move || drain_results(result_rx, drain_scheduler, drain_callbacks));
+    runtime.spawn_blocking(move || {
+        drain_results(result_rx, drain_scheduler, drain_callbacks, capacity)
+    });
 
     // Lifecycle loop: heartbeat until stopped, then unregister.
     spawn_lifecycle(
@@ -315,11 +317,17 @@ fn parse_rate_spec(
     }
 }
 
-/// Drain results: apply each to storage and invoke `WorkerBridge.onOutcome`.
+/// Drain results: finalize each batch in one transaction, then invoke
+/// `WorkerBridge.onOutcome` per job.
+///
+/// `max_batch` bounds one drain. Unlike the other shells, this result channel is
+/// unbounded (so the dispatcher never blocks a runtime worker), so nothing else
+/// would cap how much a single drain swallows.
 fn drain_results(
     result_rx: crossbeam_channel::Receiver<taskito_core::scheduler::JobResult>,
     scheduler: Arc<Scheduler>,
     callbacks: GlobalRef,
+    max_batch: usize,
 ) {
     let vm = jvm::vm();
     let mut env = match vm.attach_current_thread() {
@@ -329,25 +337,40 @@ fn drain_results(
             return;
         }
     };
-    while let Ok(result) = result_rx.recv() {
-        let outcome = match scheduler.handle_result(result) {
-            Ok(outcome) => outcome,
-            Err(e) => {
-                log::error!("[taskito-java] result handling error: {e}");
-                continue;
+    while let Ok(first) = result_rx.recv() {
+        // Finalize everything already queued in one transaction rather than one
+        // per wake.
+        let mut batch = vec![first];
+        while batch.len() < max_batch {
+            match result_rx.try_recv() {
+                Ok(more) => batch.push(more),
+                Err(_) => break,
             }
-        };
-        // Each outcome allocates several JNI locals on this long-lived attached
-        // env; scope them in a frame so a busy worker can't exhaust the
-        // local-reference table over the loop's lifetime.
-        let framed = env.with_local_frame(16, |env| {
-            if let Err(e) = call_on_outcome(env, &callbacks, &outcome) {
-                log::error!("[taskito-java] onOutcome failed: {e}");
+        }
+
+        // One outcome per input, in order, so each job is reported exactly once.
+        for outcome in scheduler.handle_results(batch) {
+            let outcome = match outcome {
+                Ok(outcome) => outcome,
+                Err(e) => {
+                    log::error!("[taskito-java] result handling error: {e}");
+                    continue;
+                }
+            };
+            // Each outcome allocates several JNI locals on this long-lived attached
+            // env; scope them in a frame so a busy worker can't exhaust the
+            // local-reference table over the loop's lifetime. The frame stays per
+            // outcome — sized for one `call_on_outcome`, it would overflow if it
+            // wrapped the whole batch.
+            let framed = env.with_local_frame(16, |env| {
+                if let Err(e) = call_on_outcome(env, &callbacks, &outcome) {
+                    log::error!("[taskito-java] onOutcome failed: {e}");
+                }
+                Ok::<(), jni::errors::Error>(())
+            });
+            if let Err(e) = framed {
+                log::error!("[taskito-java] drain local frame failed: {e}");
             }
-            Ok::<(), jni::errors::Error>(())
-        });
-        if let Err(e) = framed {
-            log::error!("[taskito-java] drain local frame failed: {e}");
         }
     }
 }
diff --git a/sdks/java/src/test/java/org/byteveda/taskito/core/ResultBatchingTest.java b/sdks/java/src/test/java/org/byteveda/taskito/core/ResultBatchingTest.java
new file mode 100644
index 00000000..eadcfb0a
--- /dev/null
+++ b/sdks/java/src/test/java/org/byteveda/taskito/core/ResultBatchingTest.java
@@ -0,0 +1,76 @@
+package org.byteveda.taskito.core;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.nio.file.Path;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import org.byteveda.taskito.Taskito;
+import org.byteveda.taskito.events.EventName;
+import org.byteveda.taskito.task.Task;
+import org.byteveda.taskito.worker.Worker;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.api.io.TempDir;
+
+/** Results finalized as a batch still surface exactly one outcome per job. */
+class ResultBatchingTest {
+
+    @Test
+    @Timeout(30)
+    void mixedBatchReportsEveryJobOnce(@TempDir Path dir) throws Exception {
+        // The drain loop finalizes everything already queued in one transaction.
+        // Successes batch into a single write while failures stay on the
+        // per-result path, so a mixed burst exercises both halves: every job must
+        // surface exactly one outcome, none lost and none double-counted.
+        int each = 6;
+        Task<String> ok = Task.of("ok", String.class);
+        Task<String> boom = Task.of("boom", String.class).maxRetries(0);
+
+        ConcurrentLinkedQueue<String> succeeded = new ConcurrentLinkedQueue<>();
+        ConcurrentLinkedQueue<String> dead = new ConcurrentLinkedQueue<>();
+        CountDownLatch finished = new CountDownLatch(each * 2);
+
+        try (Taskito queue = Taskito.builder()
+                .backend("sqlite")
+                .url(dir.resolve("t.db").toString())
+                .open()) {
+            for (int i = 0; i < each; i++) {
+                queue.enqueue(ok, String.valueOf(i));
+                queue.enqueue(boom, String.valueOf(i));
+            }
+
+            try (Worker worker = queue.worker()
+                    .concurrency(8)
+                    .batchSize(each * 2)
+                    .handle(ok, (String p) -> p)
+                    .handle(boom, (String p) -> {
+                        throw new IllegalStateException("expected");
+                    })
+                    .on(EventName.SUCCESS, event -> {
+                        succeeded.add(event.jobId);
+                        finished.countDown();
+                    })
+                    .on(EventName.DEAD, event -> {
+                        dead.add(event.jobId);
+                        finished.countDown();
+                    })
+                    .start()) {
+                assertTrue(finished.await(20, TimeUnit.SECONDS), "every job should report an outcome");
+            }
+        }
+
+        List<String> succeededIds = List.copyOf(succeeded);
+        List<String> deadIds = List.copyOf(dead);
+        assertEquals(each, succeededIds.size(), "one success outcome per successful job");
+        assertEquals(each, deadIds.size(), "one dead outcome per failing job");
+        assertEquals(each, succeededIds.stream().collect(Collectors.toSet()).size(), "no job may be reported twice");
+        Set<String> overlap = succeededIds.stream().filter(deadIds::contains).collect(Collectors.toSet());
+        assertTrue(overlap.isEmpty(), "a job cannot both succeed and dead-letter: " + overlap);
+    }
+}

From 05802ab491a59c027362a446ef63872417ef81f3 Mon Sep 17 00:00:00 2001
From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com>
Date: Thu, 16 Jul 2026 11:46:55 +0530
Subject: [PATCH 06/13] feat(node): surface retryBudget
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Caps how fast a task may retry across all of its jobs; once spent, failures
dead-letter rather than retrying. Landed for Python in #435.

Nothing was added to the config gate for it — the gate derives from the built
config, so a new option reaches the scheduler on its own. The rate parser now
names the field, since two options share its "100/m" grammar.
---
 crates/taskito-node/src/config.rs             |  3 ++
 .../taskito-node/src/convert/task_config.rs   |  4 +-
 sdks/node/src/types.ts                        |  9 ++++
 sdks/node/src/worker.ts                       |  1 +
 sdks/node/test/worker/taskConfig.test.ts      | 43 +++++++++++++++++++
 5 files changed, 58 insertions(+), 2 deletions(-)

diff --git a/crates/taskito-node/src/config.rs b/crates/taskito-node/src/config.rs
index fe27a7eb..02806571 100644
--- a/crates/taskito-node/src/config.rs
+++ b/crates/taskito-node/src/config.rs
@@ -114,6 +114,9 @@ pub struct TaskConfigInput {
     pub max_in_flight_per_task: Option<i32>,
     /// Rate-limit spec like `"100/m"`, `"50/s"`, `"3600/h"`.
     pub rate_limit: Option<String>,
+    /// Cap on how fast this task may *retry*, across all of its jobs. Same spec
+    /// as `rate_limit`; once spent, failures dead-letter instead of retrying.
+    pub retry_budget: Option<String>,
     pub circuit_breaker: Option<CircuitBreakerInput>,
 }
 
diff --git a/crates/taskito-node/src/convert/task_config.rs b/crates/taskito-node/src/convert/task_config.rs
index 5784ea8c..6f82c35a 100644
--- a/crates/taskito-node/src/convert/task_config.rs
+++ b/crates/taskito-node/src/convert/task_config.rs
@@ -42,8 +42,8 @@ pub fn task_config(input: &TaskConfigInput) -> Result<TaskConfig> {
             .as_ref()
             .map(circuit_breaker_config)
             .transpose()?,
-        // Not surfaced on this SDK yet; retries stay bounded per job.
-        retry_budget: None,
+        // Same "100/m" grammar as rate_limit, so one parser serves both.
+        retry_budget: parse_rate_spec("retryBudget", input.retry_budget.as_deref())?,
         max_concurrent: input.max_concurrent,
         max_in_flight_per_task: input.max_in_flight_per_task.map(|n| n.max(1) as usize),
     })
diff --git a/sdks/node/src/types.ts b/sdks/node/src/types.ts
index 9dbf07b7..80db1fb6 100644
--- a/sdks/node/src/types.ts
+++ b/sdks/node/src/types.ts
@@ -109,6 +109,15 @@ export interface TaskOptions {
   maxInFlightPerTask?: number;
   /** Rate-limit spec like `"100/m"`, `"50/s"`, `"3600/h"`. */
   rateLimit?: RateLimit;
+  /**
+   * Cap on how fast this task may **retry**, across all of its jobs — same spec
+   * as {@link TaskOptions.rateLimit}. Once spent, failures dead-letter instead
+   * of retrying, so a broken dependency cannot become a retry storm. Distinct
+   * from {@link TaskOptions.maxRetries}, which bounds one job rather than the
+   * rate, and from {@link TaskOptions.circuitBreaker}, which trips on hard
+   * failure rather than aggregate retry rate.
+   */
+  retryBudget?: RateLimit;
   /** Trip the task's circuit breaker after repeated failures. */
   circuitBreaker?: CircuitBreakerInput;
   /**
diff --git a/sdks/node/src/worker.ts b/sdks/node/src/worker.ts
index 330b50ca..586882dd 100644
--- a/sdks/node/src/worker.ts
+++ b/sdks/node/src/worker.ts
@@ -333,6 +333,7 @@ function toTaskConfig(name: string, options: TaskOptions): TaskConfigInput {
     maxConcurrent: options.maxConcurrent,
     maxInFlightPerTask: options.maxInFlightPerTask,
     rateLimit: options.rateLimit,
+    retryBudget: options.retryBudget,
     circuitBreaker: options.circuitBreaker,
   };
 }
diff --git a/sdks/node/test/worker/taskConfig.test.ts b/sdks/node/test/worker/taskConfig.test.ts
index 54bdfb8c..42a53feb 100644
--- a/sdks/node/test/worker/taskConfig.test.ts
+++ b/sdks/node/test/worker/taskConfig.test.ts
@@ -15,6 +15,21 @@ function newQueue(): Queue {
   return new Queue({ dbPath: join(mkdtempSync(join(tmpdir(), "taskito-tc-")), "q.db") });
 }
 
+/** Poll an async predicate. `waitFor` takes a sync one — a promise is always truthy there. */
+async function waitForAsync(
+  predicate: () => Promise<boolean>,
+  timeoutMs = 10000,
+): Promise<boolean> {
+  const deadline = Date.now() + timeoutMs;
+  while (Date.now() < deadline) {
+    if (await predicate()) {
+      return true;
+    }
+    await new Promise((resolve) => setTimeout(resolve, 20));
+  }
+  return false;
+}
+
 async function waitFor(predicate: () => boolean, timeoutMs = 10000): Promise<boolean> {
   const deadline = Date.now() + timeoutMs;
   while (Date.now() < deadline) {
@@ -65,3 +80,31 @@ it("rejects a malformed rateLimit rather than silently disabling it", async () =
   queue.task("bad", async () => {}, { rateLimit: "100/mm" as never });
   expect(() => queue.runWorker({ queues: ["default"] })).toThrow(/rateLimit/);
 });
+
+it("registers a task that sets only retryBudget", async () => {
+  // Nothing was added to the config gate for this option — it derives from the
+  // built config — so this asserts a task setting only retryBudget still reaches
+  // the scheduler. One token means the second failure dead-letters rather than
+  // retrying, even though maxRetries would allow more.
+  const queue = newQueue();
+
+  queue.task(
+    "flaky",
+    async () => {
+      throw new Error("dependency down");
+    },
+    { retryBudget: "1/m", maxRetries: 5 },
+  );
+
+  for (let i = 0; i < 3; i += 1) {
+    await queue.enqueue("flaky");
+  }
+
+  worker = queue.runWorker({ queues: ["default"], concurrency: 4, batchSize: 3 });
+
+  const budgetKilled = async (): Promise<number> => {
+    const dead = await queue.deadLetters(10);
+    return dead.filter((entry) => entry.metadata === "retry_budget_exhausted").length;
+  };
+  expect(await waitForAsync(async () => (await budgetKilled()) > 0)).toBe(true);
+});

From dea898c248706955c57b014aede03a055f70b3f5 Mon Sep 17 00:00:00 2001
From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com>
Date: Thu, 16 Jul 2026 11:47:04 +0530
Subject: [PATCH 07/13] feat(java): surface retryBudget

Caps how fast a task may retry across all of its jobs; once spent, failures
dead-letter rather than retrying. Landed for Python in #435.

On Task and @TaskHandler beside the rate limit, sharing its parser and its
fail-fast on a malformed spec. Completes this SDK's per-task policy surface.
---
 crates/taskito-java/src/convert.rs            |   3 +
 crates/taskito-java/src/worker.rs             |   5 +-
 .../processor/TaskHandlerProcessor.java       |   4 +
 .../taskito/annotation/TaskHandler.java       |   7 ++
 .../java/org/byteveda/taskito/task/Task.java  | 101 ++++++++++++++++--
 .../org/byteveda/taskito/worker/Worker.java   |   4 +
 .../taskito/core/TaskPolicyConfigTest.java    |  34 ++++++
 7 files changed, 148 insertions(+), 10 deletions(-)

diff --git a/crates/taskito-java/src/convert.rs b/crates/taskito-java/src/convert.rs
index a520ba2b..f5eee28f 100644
--- a/crates/taskito-java/src/convert.rs
+++ b/crates/taskito-java/src/convert.rs
@@ -338,6 +338,9 @@ pub struct TaskRetryConfig {
     pub circuit_breaker_half_open_success_rate: Option<f64>,
     /// Rate-limit spec like `"100/m"`, `"50/s"`, `"3600/h"`.
     pub rate_limit: Option<String>,
+    /// Cap on how fast this task may *retry*, across all of its jobs. Same spec
+    /// as `rate_limit`; once spent, failures dead-letter instead of retrying.
+    pub retry_budget: Option<String>,
     /// Cap on concurrently-running jobs of this task, across the cluster.
     pub max_concurrent: Option<i32>,
 }
diff --git a/crates/taskito-java/src/worker.rs b/crates/taskito-java/src/worker.rs
index 7a06d044..8b6da1bf 100644
--- a/crates/taskito-java/src/worker.rs
+++ b/crates/taskito-java/src/worker.rs
@@ -283,14 +283,15 @@ fn register_task_policies(
                         .unwrap_or(0.8),
                 });
         let rate_limit = parse_rate_spec("rateLimit", &config.name, config.rate_limit.as_deref())?;
+        let retry_budget =
+            parse_rate_spec("retryBudget", &config.name, config.retry_budget.as_deref())?;
         scheduler.register_task(
             config.name,
             TaskConfig {
                 retry_policy,
                 rate_limit,
                 circuit_breaker,
-                // Not surfaced on this SDK yet; retries stay bounded per job.
-                retry_budget: None,
+                retry_budget,
                 max_concurrent: config.max_concurrent,
                 // Not surfaced on this SDK yet; the whole pool stays available.
                 max_in_flight_per_task: None,
diff --git a/sdks/java/processor/src/main/java/org/byteveda/taskito/processor/TaskHandlerProcessor.java b/sdks/java/processor/src/main/java/org/byteveda/taskito/processor/TaskHandlerProcessor.java
index e82bdf54..1d4e7ad1 100644
--- a/sdks/java/processor/src/main/java/org/byteveda/taskito/processor/TaskHandlerProcessor.java
+++ b/sdks/java/processor/src/main/java/org/byteveda/taskito/processor/TaskHandlerProcessor.java
@@ -202,6 +202,10 @@ private String options(ExecutableElement method) {
         if (!rateLimit.isEmpty()) {
             chain.append(".rateLimit(\"").append(rateLimit).append("\")");
         }
+        String retryBudget = stringValue(mirror, "retryBudget", "");
+        if (!retryBudget.isEmpty()) {
+            chain.append(".retryBudget(\"").append(retryBudget).append("\")");
+        }
         long maxConcurrent = longValue(mirror, "maxConcurrent", 0);
         if (maxConcurrent > 0) {
             chain.append(".maxConcurrent(").append(maxConcurrent).append(")");
diff --git a/sdks/java/src/main/java/org/byteveda/taskito/annotation/TaskHandler.java b/sdks/java/src/main/java/org/byteveda/taskito/annotation/TaskHandler.java
index 6fe3a170..d9d95fa6 100644
--- a/sdks/java/src/main/java/org/byteveda/taskito/annotation/TaskHandler.java
+++ b/sdks/java/src/main/java/org/byteveda/taskito/annotation/TaskHandler.java
@@ -43,6 +43,13 @@
      */
     String rateLimit() default "";
 
+    /**
+     * Cap on how fast this task may <em>retry</em>, across all of its jobs — a
+     * spec like {@code "100/m"}; empty (default) leaves retries uncapped. Once
+     * spent, failures dead-letter instead of retrying.
+     */
+    String retryBudget() default "";
+
     /**
      * Cap on concurrently-running jobs of this task across the cluster; 0
      * (default) leaves it uncapped.
diff --git a/sdks/java/src/main/java/org/byteveda/taskito/task/Task.java b/sdks/java/src/main/java/org/byteveda/taskito/task/Task.java
index 4460492b..3ca6ffef 100644
--- a/sdks/java/src/main/java/org/byteveda/taskito/task/Task.java
+++ b/sdks/java/src/main/java/org/byteveda/taskito/task/Task.java
@@ -23,6 +23,7 @@ public final class Task<T> {
     private final boolean idempotent;
     private final CircuitBreakerConfig circuitBreaker;
     private final String rateLimit;
+    private final String retryBudget;
     private final Integer maxConcurrent;
 
     private Task(
@@ -34,6 +35,7 @@ private Task(
             boolean idempotent,
             CircuitBreakerConfig circuitBreaker,
             String rateLimit,
+            String retryBudget,
             Integer maxConcurrent) {
         this.name = Objects.requireNonNull(name, "task name must not be null");
         if (name.trim().isEmpty()) {
@@ -46,23 +48,34 @@ private Task(
         this.idempotent = idempotent;
         this.circuitBreaker = circuitBreaker;
         this.rateLimit = rateLimit;
+        this.retryBudget = retryBudget;
         this.maxConcurrent = maxConcurrent;
     }
 
     /** A task whose payload deserializes to {@code payloadType}. */
     public static <T> Task<T> of(String name, Class<T> payloadType) {
-        return new Task<>(name, payloadType, EnqueueOptions.none(), null, List.of(), false, null, null, null);
+        return new Task<>(name, payloadType, EnqueueOptions.none(), null, List.of(), false, null, null, null, null);
     }
 
     /** A task whose payload deserializes to a generic type, e.g. {@code new TypeReference<List<Foo>>(){}}. */
     public static <T> Task<T> of(String name, TypeReference<T> payloadType) {
-        return new Task<>(name, payloadType.getType(), EnqueueOptions.none(), null, List.of(), false, null, null, null);
+        return new Task<>(
+                name, payloadType.getType(), EnqueueOptions.none(), null, List.of(), false, null, null, null, null);
     }
 
     /** A copy of this task with the given default options. */
     public Task<T> withOptions(EnqueueOptions options) {
         return new Task<>(
-                name, payloadType, options, retryPolicy, codecs, idempotent, circuitBreaker, rateLimit, maxConcurrent);
+                name,
+                payloadType,
+                options,
+                retryPolicy,
+                codecs,
+                idempotent,
+                circuitBreaker,
+                rateLimit,
+                retryBudget,
+                maxConcurrent);
     }
 
     /**
@@ -72,7 +85,16 @@ public Task<T> withOptions(EnqueueOptions options) {
      */
     public Task<T> retryPolicy(RetryPolicy retryPolicy) {
         return new Task<>(
-                name, payloadType, options, retryPolicy, codecs, idempotent, circuitBreaker, rateLimit, maxConcurrent);
+                name,
+                payloadType,
+                options,
+                retryPolicy,
+                codecs,
+                idempotent,
+                circuitBreaker,
+                rateLimit,
+                retryBudget,
+                maxConcurrent);
     }
 
     /**
@@ -91,6 +113,7 @@ public Task<T> codecs(String... codecs) {
                 idempotent,
                 circuitBreaker,
                 rateLimit,
+                retryBudget,
                 maxConcurrent);
     }
 
@@ -101,7 +124,16 @@ public Task<T> codecs(String... codecs) {
      */
     public Task<T> idempotent(boolean idempotent) {
         return new Task<>(
-                name, payloadType, options, retryPolicy, codecs, idempotent, circuitBreaker, rateLimit, maxConcurrent);
+                name,
+                payloadType,
+                options,
+                retryPolicy,
+                codecs,
+                idempotent,
+                circuitBreaker,
+                rateLimit,
+                retryBudget,
+                maxConcurrent);
     }
 
     /**
@@ -111,7 +143,16 @@ public Task<T> idempotent(boolean idempotent) {
      */
     public Task<T> circuitBreaker(CircuitBreakerConfig circuitBreaker) {
         return new Task<>(
-                name, payloadType, options, retryPolicy, codecs, idempotent, circuitBreaker, rateLimit, maxConcurrent);
+                name,
+                payloadType,
+                options,
+                retryPolicy,
+                codecs,
+                idempotent,
+                circuitBreaker,
+                rateLimit,
+                retryBudget,
+                maxConcurrent);
     }
 
     /**
@@ -121,7 +162,16 @@ public Task<T> circuitBreaker(CircuitBreakerConfig circuitBreaker) {
      */
     public Task<T> rateLimit(String rateLimit) {
         return new Task<>(
-                name, payloadType, options, retryPolicy, codecs, idempotent, circuitBreaker, rateLimit, maxConcurrent);
+                name,
+                payloadType,
+                options,
+                retryPolicy,
+                codecs,
+                idempotent,
+                circuitBreaker,
+                rateLimit,
+                retryBudget,
+                maxConcurrent);
     }
 
     /**
@@ -131,7 +181,37 @@ public Task<T> rateLimit(String rateLimit) {
      */
     public Task<T> maxConcurrent(Integer maxConcurrent) {
         return new Task<>(
-                name, payloadType, options, retryPolicy, codecs, idempotent, circuitBreaker, rateLimit, maxConcurrent);
+                name,
+                payloadType,
+                options,
+                retryPolicy,
+                codecs,
+                idempotent,
+                circuitBreaker,
+                rateLimit,
+                retryBudget,
+                maxConcurrent);
+    }
+
+    /**
+     * A copy of this task whose <em>retries</em> are capped at {@code retryBudget},
+     * a spec like {@code "100/m"} — across all of its jobs, not per job. Once spent,
+     * failures dead-letter instead of retrying, so a broken dependency cannot become
+     * a retry storm. Distinct from {@link #maxRetries}, which bounds one job rather
+     * than the rate.
+     */
+    public Task<T> retryBudget(String retryBudget) {
+        return new Task<>(
+                name,
+                payloadType,
+                options,
+                retryPolicy,
+                codecs,
+                idempotent,
+                circuitBreaker,
+                rateLimit,
+                retryBudget,
+                maxConcurrent);
     }
 
     public Task<T> queue(String queue) {
@@ -205,6 +285,11 @@ public String rateLimit() {
         return rateLimit;
     }
 
+    /** This task's retry-rate cap (e.g. {@code "100/m"}), or {@code null} when uncapped. */
+    public String retryBudget() {
+        return retryBudget;
+    }
+
     /** Cap on this task's concurrently-running jobs, or {@code null} when uncapped. */
     public Integer maxConcurrent() {
         return maxConcurrent;
diff --git a/sdks/java/src/main/java/org/byteveda/taskito/worker/Worker.java b/sdks/java/src/main/java/org/byteveda/taskito/worker/Worker.java
index 1a060e65..1e0b3b20 100644
--- a/sdks/java/src/main/java/org/byteveda/taskito/worker/Worker.java
+++ b/sdks/java/src/main/java/org/byteveda/taskito/worker/Worker.java
@@ -252,6 +252,7 @@ private static boolean hasPolicy(Task<?> task) {
             return task.retryPolicy() != null
                     || task.circuitBreaker() != null
                     || task.rateLimit() != null
+                    || task.retryBudget() != null
                     || task.maxConcurrent() != null;
         }
 
@@ -497,6 +498,9 @@ private List<Map<String, Object>> encodeTaskConfigs() {
                 if (task.rateLimit() != null) {
                     config.put("rateLimit", task.rateLimit());
                 }
+                if (task.retryBudget() != null) {
+                    config.put("retryBudget", task.retryBudget());
+                }
                 if (task.maxConcurrent() != null) {
                     config.put("maxConcurrent", task.maxConcurrent());
                 }
diff --git a/sdks/java/src/test/java/org/byteveda/taskito/core/TaskPolicyConfigTest.java b/sdks/java/src/test/java/org/byteveda/taskito/core/TaskPolicyConfigTest.java
index a6b4baf9..26df5205 100644
--- a/sdks/java/src/test/java/org/byteveda/taskito/core/TaskPolicyConfigTest.java
+++ b/sdks/java/src/test/java/org/byteveda/taskito/core/TaskPolicyConfigTest.java
@@ -91,6 +91,40 @@ void rateLimitThrottlesDispatch(@TempDir Path dir) throws Exception {
         }
     }
 
+    /** The budget caps retries across jobs, so a storm dead-letters rather than retrying forever. */
+    @Test
+    @Timeout(30)
+    void retryBudgetDeadLettersOnceSpent(@TempDir Path dir) throws Exception {
+        // maxRetries(5) would allow plenty of retries per job; one budget token
+        // across all of them means the rest dead-letter instead.
+        Task<String> flaky = Task.of("flaky", String.class).retryBudget("1/m").maxRetries(5);
+        CountDownLatch dead = new CountDownLatch(1);
+
+        try (Taskito queue = Taskito.builder()
+                .backend("sqlite")
+                .url(dir.resolve("t.db").toString())
+                .open()) {
+            for (int i = 0; i < 3; i++) {
+                queue.enqueue(flaky, String.valueOf(i));
+            }
+
+            try (Worker worker = queue.worker()
+                    .concurrency(4)
+                    .handle(flaky, (String p) -> {
+                        throw new IllegalStateException("dependency down");
+                    })
+                    .on(EventName.DEAD, event -> dead.countDown())
+                    .start()) {
+                assertTrue(dead.await(20, TimeUnit.SECONDS), "an over-budget job should dead-letter");
+            }
+
+            long budgetKilled = queue.listDead(10, 0).stream()
+                    .filter(entry -> "retry_budget_exhausted".equals(entry.metadata))
+                    .count();
+            assertTrue(budgetKilled > 0, "a job should be dead-lettered by the budget, not by retry exhaustion");
+        }
+    }
+
     /** A typo must fail the start, not silently run the task unthrottled. */
     @Test
     @Timeout(30)

From 630caabbd0b70a3bc70187601f9552a860297d27 Mon Sep 17 00:00:00 2001
From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com>
Date: Thu, 16 Jul 2026 11:51:35 +0530
Subject: [PATCH 08/13] feat(java): surface maxInFlightPerTask

The last of the four per-task knobs this SDK carried no wire field for. All
three SDKs now surface every one, so no binding hardcodes a None the caller
cannot reach.
---
 crates/taskito-java/src/convert.rs            |  3 +
 crates/taskito-java/src/worker.rs             |  3 +-
 .../processor/TaskHandlerProcessor.java       |  4 ++
 .../taskito/annotation/TaskHandler.java       |  6 ++
 .../java/org/byteveda/taskito/task/Task.java  | 70 ++++++++++++++++---
 .../org/byteveda/taskito/worker/Worker.java   |  6 +-
 .../taskito/core/TaskPolicyConfigTest.java    | 41 +++++++++++
 7 files changed, 119 insertions(+), 14 deletions(-)

diff --git a/crates/taskito-java/src/convert.rs b/crates/taskito-java/src/convert.rs
index f5eee28f..f0c7416d 100644
--- a/crates/taskito-java/src/convert.rs
+++ b/crates/taskito-java/src/convert.rs
@@ -343,6 +343,9 @@ pub struct TaskRetryConfig {
     pub retry_budget: Option<String>,
     /// Cap on concurrently-running jobs of this task, across the cluster.
     pub max_concurrent: Option<i32>,
+    /// Cap on this task's share of one worker's dispatch slots. In-process,
+    /// unlike `max_concurrent`, which is cluster-wide and costs a DB read.
+    pub max_in_flight_per_task: Option<i32>,
 }
 
 /// Filter accepted by `NativeQueue.listJobs`.
diff --git a/crates/taskito-java/src/worker.rs b/crates/taskito-java/src/worker.rs
index 8b6da1bf..c38288d3 100644
--- a/crates/taskito-java/src/worker.rs
+++ b/crates/taskito-java/src/worker.rs
@@ -293,8 +293,7 @@ fn register_task_policies(
                 circuit_breaker,
                 retry_budget,
                 max_concurrent: config.max_concurrent,
-                // Not surfaced on this SDK yet; the whole pool stays available.
-                max_in_flight_per_task: None,
+                max_in_flight_per_task: config.max_in_flight_per_task.map(|n| n.max(1) as usize),
             },
         );
     }
diff --git a/sdks/java/processor/src/main/java/org/byteveda/taskito/processor/TaskHandlerProcessor.java b/sdks/java/processor/src/main/java/org/byteveda/taskito/processor/TaskHandlerProcessor.java
index 1d4e7ad1..d4be4596 100644
--- a/sdks/java/processor/src/main/java/org/byteveda/taskito/processor/TaskHandlerProcessor.java
+++ b/sdks/java/processor/src/main/java/org/byteveda/taskito/processor/TaskHandlerProcessor.java
@@ -210,6 +210,10 @@ private String options(ExecutableElement method) {
         if (maxConcurrent > 0) {
             chain.append(".maxConcurrent(").append(maxConcurrent).append(")");
         }
+        long maxInFlightPerTask = longValue(mirror, "maxInFlightPerTask", 0);
+        if (maxInFlightPerTask > 0) {
+            chain.append(".maxInFlightPerTask(").append(maxInFlightPerTask).append(")");
+        }
         appendCircuitBreaker(chain, mirror);
         return chain.toString();
     }
diff --git a/sdks/java/src/main/java/org/byteveda/taskito/annotation/TaskHandler.java b/sdks/java/src/main/java/org/byteveda/taskito/annotation/TaskHandler.java
index d9d95fa6..115d46cf 100644
--- a/sdks/java/src/main/java/org/byteveda/taskito/annotation/TaskHandler.java
+++ b/sdks/java/src/main/java/org/byteveda/taskito/annotation/TaskHandler.java
@@ -56,6 +56,12 @@
      */
     int maxConcurrent() default 0;
 
+    /**
+     * Cap on this task's share of one worker's dispatch slots, so a slow task
+     * cannot occupy the whole pool; 0 (default) lets it use the whole pool.
+     */
+    int maxInFlightPerTask() default 0;
+
     /** Circuit-breaker failure threshold; 0 (default) leaves the breaker off. */
     int circuitBreakerThreshold() default 0;
 
diff --git a/sdks/java/src/main/java/org/byteveda/taskito/task/Task.java b/sdks/java/src/main/java/org/byteveda/taskito/task/Task.java
index 3ca6ffef..bbf3b5c9 100644
--- a/sdks/java/src/main/java/org/byteveda/taskito/task/Task.java
+++ b/sdks/java/src/main/java/org/byteveda/taskito/task/Task.java
@@ -25,6 +25,7 @@ public final class Task<T> {
     private final String rateLimit;
     private final String retryBudget;
     private final Integer maxConcurrent;
+    private final Integer maxInFlightPerTask;
 
     private Task(
             String name,
@@ -36,7 +37,8 @@ private Task(
             CircuitBreakerConfig circuitBreaker,
             String rateLimit,
             String retryBudget,
-            Integer maxConcurrent) {
+            Integer maxConcurrent,
+            Integer maxInFlightPerTask) {
         this.name = Objects.requireNonNull(name, "task name must not be null");
         if (name.trim().isEmpty()) {
             throw new IllegalArgumentException("task name must not be blank");
@@ -50,17 +52,29 @@ private Task(
         this.rateLimit = rateLimit;
         this.retryBudget = retryBudget;
         this.maxConcurrent = maxConcurrent;
+        this.maxInFlightPerTask = maxInFlightPerTask;
     }
 
     /** A task whose payload deserializes to {@code payloadType}. */
     public static <T> Task<T> of(String name, Class<T> payloadType) {
-        return new Task<>(name, payloadType, EnqueueOptions.none(), null, List.of(), false, null, null, null, null);
+        return new Task<>(
+                name, payloadType, EnqueueOptions.none(), null, List.of(), false, null, null, null, null, null);
     }
 
     /** A task whose payload deserializes to a generic type, e.g. {@code new TypeReference<List<Foo>>(){}}. */
     public static <T> Task<T> of(String name, TypeReference<T> payloadType) {
         return new Task<>(
-                name, payloadType.getType(), EnqueueOptions.none(), null, List.of(), false, null, null, null, null);
+                name,
+                payloadType.getType(),
+                EnqueueOptions.none(),
+                null,
+                List.of(),
+                false,
+                null,
+                null,
+                null,
+                null,
+                null);
     }
 
     /** A copy of this task with the given default options. */
@@ -75,7 +89,8 @@ public Task<T> withOptions(EnqueueOptions options) {
                 circuitBreaker,
                 rateLimit,
                 retryBudget,
-                maxConcurrent);
+                maxConcurrent,
+                maxInFlightPerTask);
     }
 
     /**
@@ -94,7 +109,8 @@ public Task<T> retryPolicy(RetryPolicy retryPolicy) {
                 circuitBreaker,
                 rateLimit,
                 retryBudget,
-                maxConcurrent);
+                maxConcurrent,
+                maxInFlightPerTask);
     }
 
     /**
@@ -114,7 +130,8 @@ public Task<T> codecs(String... codecs) {
                 circuitBreaker,
                 rateLimit,
                 retryBudget,
-                maxConcurrent);
+                maxConcurrent,
+                maxInFlightPerTask);
     }
 
     /**
@@ -133,7 +150,8 @@ public Task<T> idempotent(boolean idempotent) {
                 circuitBreaker,
                 rateLimit,
                 retryBudget,
-                maxConcurrent);
+                maxConcurrent,
+                maxInFlightPerTask);
     }
 
     /**
@@ -152,7 +170,8 @@ public Task<T> circuitBreaker(CircuitBreakerConfig circuitBreaker) {
                 circuitBreaker,
                 rateLimit,
                 retryBudget,
-                maxConcurrent);
+                maxConcurrent,
+                maxInFlightPerTask);
     }
 
     /**
@@ -171,7 +190,8 @@ public Task<T> rateLimit(String rateLimit) {
                 circuitBreaker,
                 rateLimit,
                 retryBudget,
-                maxConcurrent);
+                maxConcurrent,
+                maxInFlightPerTask);
     }
 
     /**
@@ -190,7 +210,8 @@ public Task<T> maxConcurrent(Integer maxConcurrent) {
                 circuitBreaker,
                 rateLimit,
                 retryBudget,
-                maxConcurrent);
+                maxConcurrent,
+                maxInFlightPerTask);
     }
 
     /**
@@ -211,7 +232,29 @@ public Task<T> retryBudget(String retryBudget) {
                 circuitBreaker,
                 rateLimit,
                 retryBudget,
-                maxConcurrent);
+                maxConcurrent,
+                maxInFlightPerTask);
+    }
+
+    /**
+     * A copy of this task allowed at most {@code maxInFlightPerTask} of one worker's
+     * dispatch slots, so a slow task cannot occupy the whole pool and starve the
+     * others. In-process and free, unlike {@link #maxConcurrent}, which is
+     * cluster-wide and costs a database read; {@code null} lets it use the whole pool.
+     */
+    public Task<T> maxInFlightPerTask(Integer maxInFlightPerTask) {
+        return new Task<>(
+                name,
+                payloadType,
+                options,
+                retryPolicy,
+                codecs,
+                idempotent,
+                circuitBreaker,
+                rateLimit,
+                retryBudget,
+                maxConcurrent,
+                maxInFlightPerTask);
     }
 
     public Task<T> queue(String queue) {
@@ -294,4 +337,9 @@ public String retryBudget() {
     public Integer maxConcurrent() {
         return maxConcurrent;
     }
+
+    /** Cap on this task's share of one worker's dispatch slots, or {@code null} when uncapped. */
+    public Integer maxInFlightPerTask() {
+        return maxInFlightPerTask;
+    }
 }
diff --git a/sdks/java/src/main/java/org/byteveda/taskito/worker/Worker.java b/sdks/java/src/main/java/org/byteveda/taskito/worker/Worker.java
index 1e0b3b20..771757ed 100644
--- a/sdks/java/src/main/java/org/byteveda/taskito/worker/Worker.java
+++ b/sdks/java/src/main/java/org/byteveda/taskito/worker/Worker.java
@@ -253,7 +253,8 @@ private static boolean hasPolicy(Task<?> task) {
                     || task.circuitBreaker() != null
                     || task.rateLimit() != null
                     || task.retryBudget() != null
-                    || task.maxConcurrent() != null;
+                    || task.maxConcurrent() != null
+                    || task.maxInFlightPerTask() != null;
         }
 
         /** Register every handler in a {@link HandlerRegistry} (e.g. a generated {@code XxxTasks.handlers}). */
@@ -504,6 +505,9 @@ private List<Map<String, Object>> encodeTaskConfigs() {
                 if (task.maxConcurrent() != null) {
                     config.put("maxConcurrent", task.maxConcurrent());
                 }
+                if (task.maxInFlightPerTask() != null) {
+                    config.put("maxInFlightPerTask", task.maxInFlightPerTask());
+                }
                 configs.add(config);
             }
             return configs;
diff --git a/sdks/java/src/test/java/org/byteveda/taskito/core/TaskPolicyConfigTest.java b/sdks/java/src/test/java/org/byteveda/taskito/core/TaskPolicyConfigTest.java
index 26df5205..a5ccf067 100644
--- a/sdks/java/src/test/java/org/byteveda/taskito/core/TaskPolicyConfigTest.java
+++ b/sdks/java/src/test/java/org/byteveda/taskito/core/TaskPolicyConfigTest.java
@@ -58,6 +58,47 @@ void maxConcurrentSerializesOneTask(@TempDir Path dir) throws Exception {
         assertEquals(1, peak.get(), "maxConcurrent=1 must never let two jobs of the task overlap");
     }
 
+    /** The in-process bulkhead holds a task to its share of one worker's slots. */
+    @Test
+    @Timeout(30)
+    void maxInFlightPerTaskBoundsOneTask(@TempDir Path dir) throws Exception {
+        // Same visible effect as maxConcurrent=1 here, but enforced in-process
+        // from the worker's in-flight map rather than a running-count query.
+        Task<String> solo = Task.of("solo", String.class).maxInFlightPerTask(1);
+        AtomicInteger running = new AtomicInteger();
+        AtomicInteger peak = new AtomicInteger();
+        CountDownLatch done = new CountDownLatch(4);
+
+        try (Taskito queue = Taskito.builder()
+                .backend("sqlite")
+                .url(dir.resolve("t.db").toString())
+                .open()) {
+            for (int i = 0; i < 4; i++) {
+                queue.enqueue(solo, String.valueOf(i));
+            }
+
+            try (Worker worker = queue.worker()
+                    .concurrency(4)
+                    .batchSize(4)
+                    .handle(solo, (String p) -> {
+                        peak.accumulateAndGet(running.incrementAndGet(), Math::max);
+                        try {
+                            Thread.sleep(150);
+                        } catch (InterruptedException e) {
+                            Thread.currentThread().interrupt();
+                        }
+                        running.decrementAndGet();
+                        return null;
+                    })
+                    .on(EventName.SUCCESS, event -> done.countDown())
+                    .start()) {
+                assertTrue(done.await(20, TimeUnit.SECONDS), "all four jobs should finish");
+            }
+        }
+
+        assertEquals(1, peak.get(), "maxInFlightPerTask=1 must never let two jobs of the task overlap");
+    }
+
     /** Throttling drags a burst out past the window it would otherwise finish in. */
     @Test
     @Timeout(30)

From ecf363df7ff4c874b40979f19b697d1c57a1971f Mon Sep 17 00:00:00 2001
From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com>
Date: Thu, 16 Jul 2026 16:29:33 +0530
Subject: [PATCH 09/13] fix(java): validate task policies before registering
 worker state
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

A malformed rate spec was rejected after the worker row and its subscriptions
were already written, and the start then returned without a handle — so the
lifecycle loop that removes them never ran and both rows leaked.

Build the policies first: parsing is pure, so a rejected config now leaves
nothing behind.
---
 crates/taskito-java/src/worker.rs | 30 ++++++++++++++++++++----------
 1 file changed, 20 insertions(+), 10 deletions(-)

diff --git a/crates/taskito-java/src/worker.rs b/crates/taskito-java/src/worker.rs
index c38288d3..9c36045b 100644
--- a/crates/taskito-java/src/worker.rs
+++ b/crates/taskito-java/src/worker.rs
@@ -92,6 +92,12 @@ fn start_worker(
     #[cfg(feature = "mesh")]
     let mesh_worker_id = worker_id.clone();
 
+    // Validate every task policy before writing any persistent state: this is
+    // the last fallible step that has no side effects, and returning Err past
+    // the writes below would leave a worker row and its subscriptions behind
+    // with no handle to run the lifecycle loop that cleans them up.
+    let task_policies = build_task_policies(options.task_configs)?;
+
     // Create the live worker row before its ephemeral subscriptions exist:
     // the reaper only spares owned rows whose owner is registered, so this
     // ordering (plus the core's registration grace window) keeps a concurrent
@@ -106,7 +112,9 @@ fn start_worker(
     // Claim execution under this worker's id so dead-worker recovery can
     // attribute orphaned jobs (matches the worker id registered above).
     scheduler.set_claim_owner(worker_id.clone());
-    register_task_policies(&mut scheduler, options.task_configs)?;
+    for (name, policy) in task_policies {
+        scheduler.register_task(name, policy);
+    }
     let scheduler = Arc::new(scheduler);
     let shutdown = scheduler.shutdown_handle();
     let heartbeat_stop = Arc::new(Notify::new());
@@ -237,19 +245,21 @@ fn register_subscriptions(
     Ok(())
 }
 
-/// Register each task's policy — retry curve, throttling, concurrency caps —
-/// with the scheduler. Unset fields keep the core's defaults; the per-job retry
+/// Build each task's policy — retry curve, throttling, concurrency caps — from
+/// its wire config. Unset fields keep the core's defaults; the per-job retry
 /// budget still resolves from the job's `max_retries`.
 ///
 /// Fails on a malformed rate spec rather than registering the task without it:
 /// silently dropping a throttle the caller asked for is worse than not starting.
-fn register_task_policies(
-    scheduler: &mut Scheduler,
+/// Pure, so the caller can validate before writing any persistent worker state
+/// and a rejected config leaves nothing behind.
+fn build_task_policies(
     configs: Option<Vec<TaskRetryConfig>>,
-) -> Result<(), crate::error::BindingError> {
+) -> Result<Vec<(String, TaskConfig)>, crate::error::BindingError> {
     let Some(configs) = configs else {
-        return Ok(());
+        return Ok(Vec::new());
     };
+    let mut built = Vec::with_capacity(configs.len());
     for config in configs {
         let mut retry_policy = RetryPolicy::default();
         if let Some(custom) = config.custom_delays_ms {
@@ -285,7 +295,7 @@ fn register_task_policies(
         let rate_limit = parse_rate_spec("rateLimit", &config.name, config.rate_limit.as_deref())?;
         let retry_budget =
             parse_rate_spec("retryBudget", &config.name, config.retry_budget.as_deref())?;
-        scheduler.register_task(
+        built.push((
             config.name,
             TaskConfig {
                 retry_policy,
@@ -295,9 +305,9 @@ fn register_task_policies(
                 max_concurrent: config.max_concurrent,
                 max_in_flight_per_task: config.max_in_flight_per_task.map(|n| n.max(1) as usize),
             },
-        );
+        ));
     }
-    Ok(())
+    Ok(built)
 }
 
 /// Parse an optional rate spec, naming the offending task and option so a typo

From 4cf8e64021c50620e0604b63ab2f54488b2292dc Mon Sep 17 00:00:00 2001
From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com>
Date: Thu, 16 Jul 2026 16:29:35 +0530
Subject: [PATCH 10/13] fix(node): cap the result batch explicitly

A bounded channel caps how many results sit in it at once, not how many one
drain takes: senders refill slots while try_recv runs, so the loop could
swallow a whole backlog into one Vec and stall finalization behind it. Cap it
like the Java drain does.
---
 crates/taskito-node/src/worker.rs | 17 ++++++++++++-----
 1 file changed, 12 insertions(+), 5 deletions(-)

diff --git a/crates/taskito-node/src/worker.rs b/crates/taskito-node/src/worker.rs
index 83126d1f..ca713fdc 100644
--- a/crates/taskito-node/src/worker.rs
+++ b/crates/taskito-node/src/worker.rs
@@ -199,12 +199,19 @@ pub fn start_worker(
     let scheduler_results = scheduler;
     spawn_blocking(move || {
         while let Ok(first) = result_rx.recv() {
-            // Take everything already queued and finalize it in one batched
-            // transaction rather than one per wake. The channel is bounded, so
-            // the drain caps itself.
+            // Finalize everything already queued in one batched transaction
+            // rather than one per wake.
+            //
+            // Capped explicitly: the channel bounds how many results can sit in
+            // it at once, not how many this loop drains — senders refill slots
+            // while `try_recv` runs, so an unbounded drain could swallow a whole
+            // backlog into one Vec and stall finalization behind it.
             let mut batch = vec![first];
-            while let Ok(more) = result_rx.try_recv() {
-                batch.push(more);
+            while batch.len() < capacity {
+                match result_rx.try_recv() {
+                    Ok(more) => batch.push(more),
+                    Err(_) => break,
+                }
             }
 
             // A panicking batch must not kill the drain loop — a dead loop

From 7457e9e3a4fabbefe0bf1f12ff123f4e53618d3d Mon Sep 17 00:00:00 2001
From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com>
Date: Thu, 16 Jul 2026 16:29:37 +0530
Subject: [PATCH 11/13] fix(sdks): name the task in rate errors, treat a zero
 cap as uncapped
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

A malformed spec on Node named the option but not the task, so a start-up
failure could not say which config was wrong. Java already named both.

Task.maxConcurrent(0)/maxInFlightPerTask(0) stored the zero verbatim, and the
scheduler reads a zero cap as "never dispatch". Zero is the annotation's unset
sentinel, so the fluent setters now read it the same way.

Also awaits the enqueues in the batching test — they could still be pending
when the worker started — and asserts the terminal ids are disjoint, which
per-list uniqueness would miss.
---
 .../taskito-node/src/convert/task_config.rs   | 21 +++++++++++--------
 .../java/org/byteveda/taskito/task/Task.java  | 14 +++++++++++--
 sdks/node/test/worker/resultBatching.test.ts  |  7 +++++--
 3 files changed, 29 insertions(+), 13 deletions(-)

diff --git a/crates/taskito-node/src/convert/task_config.rs b/crates/taskito-node/src/convert/task_config.rs
index 6f82c35a..69a6031c 100644
--- a/crates/taskito-node/src/convert/task_config.rs
+++ b/crates/taskito-node/src/convert/task_config.rs
@@ -16,13 +16,16 @@ const DEFAULT_HALF_OPEN_PROBES: i32 = 5;
 const DEFAULT_HALF_OPEN_SUCCESS_RATE: f64 = 0.8;
 
 /// Parse an optional rate spec, failing fast on a malformed value rather than
-/// silently disabling the cap (a misconfigured limit is a config error).
-/// `field` names the option in the error, since several share this grammar.
-fn parse_rate_spec(field: &str, spec: Option<&str>) -> Result<Option<RateLimitConfig>> {
+/// silently disabling the cap (a misconfigured limit is a config error). Names
+/// the option and the task, since several options share this grammar and the
+/// error otherwise cannot say which config is wrong.
+fn parse_rate_spec(field: &str, task: &str, spec: Option<&str>) -> Result<Option<RateLimitConfig>> {
     match spec {
-        Some(s) => RateLimitConfig::parse(s)
-            .map(Some)
-            .ok_or_else(|| invalid_arg(format!("invalid {field} '{s}' (expected e.g. '100/m')"))),
+        Some(s) => RateLimitConfig::parse(s).map(Some).ok_or_else(|| {
+            invalid_arg(format!(
+                "invalid {field} '{s}' on task '{task}' (expected e.g. '100/m')"
+            ))
+        }),
         None => Ok(None),
     }
 }
@@ -36,14 +39,14 @@ pub fn task_config(input: &TaskConfigInput) -> Result<TaskConfig> {
             max_delay_ms: input.retry_max_delay_ms.unwrap_or(DEFAULT_RETRY_MAX_MS),
             custom_delays_ms: None,
         },
-        rate_limit: parse_rate_spec("rateLimit", input.rate_limit.as_deref())?,
+        rate_limit: parse_rate_spec("rateLimit", &input.name, input.rate_limit.as_deref())?,
         circuit_breaker: input
             .circuit_breaker
             .as_ref()
             .map(circuit_breaker_config)
             .transpose()?,
         // Same "100/m" grammar as rate_limit, so one parser serves both.
-        retry_budget: parse_rate_spec("retryBudget", input.retry_budget.as_deref())?,
+        retry_budget: parse_rate_spec("retryBudget", &input.name, input.retry_budget.as_deref())?,
         max_concurrent: input.max_concurrent,
         max_in_flight_per_task: input.max_in_flight_per_task.map(|n| n.max(1) as usize),
     })
@@ -88,7 +91,7 @@ fn circuit_breaker_config(input: &CircuitBreakerInput) -> Result<CircuitBreakerC
 /// Build a [`QueueConfig`] (rate limit, concurrency cap) from JS input.
 pub fn queue_config(input: &QueueConfigInput) -> Result<QueueConfig> {
     Ok(QueueConfig {
-        rate_limit: parse_rate_spec("rateLimit", input.rate_limit.as_deref())?,
+        rate_limit: parse_rate_spec("rateLimit", &input.name, input.rate_limit.as_deref())?,
         max_concurrent: input.max_concurrent,
     })
 }
diff --git a/sdks/java/src/main/java/org/byteveda/taskito/task/Task.java b/sdks/java/src/main/java/org/byteveda/taskito/task/Task.java
index bbf3b5c9..b429ea87 100644
--- a/sdks/java/src/main/java/org/byteveda/taskito/task/Task.java
+++ b/sdks/java/src/main/java/org/byteveda/taskito/task/Task.java
@@ -197,9 +197,12 @@ public Task<T> rateLimit(String rateLimit) {
     /**
      * A copy of this task allowed at most {@code maxConcurrent} jobs running at once
      * across the cluster. The scheduler counts running jobs before dispatch, so this
-     * costs a database read; {@code null} means no cap.
+     * costs a database read. {@code null} or {@code 0} means no cap — matching the
+     * annotation's sentinel, and because a literal cap of zero would stop the task
+     * from ever dispatching.
      */
     public Task<T> maxConcurrent(Integer maxConcurrent) {
+        maxConcurrent = uncappedIfZero(maxConcurrent);
         return new Task<>(
                 name,
                 payloadType,
@@ -240,9 +243,11 @@ public Task<T> retryBudget(String retryBudget) {
      * A copy of this task allowed at most {@code maxInFlightPerTask} of one worker's
      * dispatch slots, so a slow task cannot occupy the whole pool and starve the
      * others. In-process and free, unlike {@link #maxConcurrent}, which is
-     * cluster-wide and costs a database read; {@code null} lets it use the whole pool.
+     * cluster-wide and costs a database read. {@code null} or {@code 0} lets it use
+     * the whole pool, matching the annotation's sentinel.
      */
     public Task<T> maxInFlightPerTask(Integer maxInFlightPerTask) {
+        maxInFlightPerTask = uncappedIfZero(maxInFlightPerTask);
         return new Task<>(
                 name,
                 payloadType,
@@ -257,6 +262,11 @@ public Task<T> maxInFlightPerTask(Integer maxInFlightPerTask) {
                 maxInFlightPerTask);
     }
 
+    /** A cap of zero is the annotation's "unset" sentinel, never a literal zero. */
+    private static Integer uncappedIfZero(Integer cap) {
+        return cap != null && cap == 0 ? null : cap;
+    }
+
     public Task<T> queue(String queue) {
         return withOptions(options.toBuilder().queue(queue).build());
     }
diff --git a/sdks/node/test/worker/resultBatching.test.ts b/sdks/node/test/worker/resultBatching.test.ts
index e5e9c6ba..bc509b00 100644
--- a/sdks/node/test/worker/resultBatching.test.ts
+++ b/sdks/node/test/worker/resultBatching.test.ts
@@ -50,8 +50,8 @@ it("reports every job exactly once when results are drained as a batch", async (
   );
 
   for (let i = 0; i < each; i += 1) {
-    queue.enqueue("ok", [i]);
-    queue.enqueue("boom");
+    await queue.enqueue("ok", [i]);
+    await queue.enqueue("boom");
   }
 
   worker = queue.runWorker({ queues: ["default"], concurrency: 8, batchSize: each * 2 });
@@ -62,4 +62,7 @@ it("reports every job exactly once when results are drained as a batch", async (
   expect(dead).toHaveLength(each);
   expect(new Set(completed).size).toBe(each);
   expect(new Set(dead).size).toBe(each);
+  // Uniqueness within each list would still miss a job reported once as both.
+  const overlap = completed.filter((id) => dead.includes(id));
+  expect(overlap).toEqual([]);
 });

From 53ae43f53527e2a8a37ab97f2881671308d9f3bd Mon Sep 17 00:00:00 2001
From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com>
Date: Thu, 16 Jul 2026 16:42:29 +0530
Subject: [PATCH 12/13] fix(node): name the config's kind, not just its name,
 in rate errors
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Queues configure a rate limit through the same parser, so wording the error
"on task" reported a malformed queue limit as a task — a regression from
adding the name in the first place. Pass the kind alongside the name.
---
 .../taskito-node/src/convert/task_config.rs   | 39 +++++++++++++++----
 sdks/node/test/worker/rateScope.test.ts       | 22 +++++++++++
 2 files changed, 53 insertions(+), 8 deletions(-)
 create mode 100644 sdks/node/test/worker/rateScope.test.ts

diff --git a/crates/taskito-node/src/convert/task_config.rs b/crates/taskito-node/src/convert/task_config.rs
index 69a6031c..e54b67fe 100644
--- a/crates/taskito-node/src/convert/task_config.rs
+++ b/crates/taskito-node/src/convert/task_config.rs
@@ -16,14 +16,22 @@ const DEFAULT_HALF_OPEN_PROBES: i32 = 5;
 const DEFAULT_HALF_OPEN_SUCCESS_RATE: f64 = 0.8;
 
 /// Parse an optional rate spec, failing fast on a malformed value rather than
-/// silently disabling the cap (a misconfigured limit is a config error). Names
-/// the option and the task, since several options share this grammar and the
-/// error otherwise cannot say which config is wrong.
-fn parse_rate_spec(field: &str, task: &str, spec: Option<&str>) -> Result<Option<RateLimitConfig>> {
+/// silently disabling the cap (a misconfigured limit is a config error).
+///
+/// Names the option and what carries it — several options share this grammar,
+/// and both tasks and queues configure one, so the error cannot say which
+/// config is wrong without both. `scope` is the kind ("task", "queue"), `name`
+/// its name.
+fn parse_rate_spec(
+    field: &str,
+    scope: &str,
+    name: &str,
+    spec: Option<&str>,
+) -> Result<Option<RateLimitConfig>> {
     match spec {
         Some(s) => RateLimitConfig::parse(s).map(Some).ok_or_else(|| {
             invalid_arg(format!(
-                "invalid {field} '{s}' on task '{task}' (expected e.g. '100/m')"
+                "invalid {field} '{s}' on {scope} '{name}' (expected e.g. '100/m')"
             ))
         }),
         None => Ok(None),
@@ -39,14 +47,24 @@ pub fn task_config(input: &TaskConfigInput) -> Result<TaskConfig> {
             max_delay_ms: input.retry_max_delay_ms.unwrap_or(DEFAULT_RETRY_MAX_MS),
             custom_delays_ms: None,
         },
-        rate_limit: parse_rate_spec("rateLimit", &input.name, input.rate_limit.as_deref())?,
+        rate_limit: parse_rate_spec(
+            "rateLimit",
+            "task",
+            &input.name,
+            input.rate_limit.as_deref(),
+        )?,
         circuit_breaker: input
             .circuit_breaker
             .as_ref()
             .map(circuit_breaker_config)
             .transpose()?,
         // Same "100/m" grammar as rate_limit, so one parser serves both.
-        retry_budget: parse_rate_spec("retryBudget", &input.name, input.retry_budget.as_deref())?,
+        retry_budget: parse_rate_spec(
+            "retryBudget",
+            "task",
+            &input.name,
+            input.retry_budget.as_deref(),
+        )?,
         max_concurrent: input.max_concurrent,
         max_in_flight_per_task: input.max_in_flight_per_task.map(|n| n.max(1) as usize),
     })
@@ -91,7 +109,12 @@ fn circuit_breaker_config(input: &CircuitBreakerInput) -> Result<CircuitBreakerC
 /// Build a [`QueueConfig`] (rate limit, concurrency cap) from JS input.
 pub fn queue_config(input: &QueueConfigInput) -> Result<QueueConfig> {
     Ok(QueueConfig {
-        rate_limit: parse_rate_spec("rateLimit", &input.name, input.rate_limit.as_deref())?,
+        rate_limit: parse_rate_spec(
+            "rateLimit",
+            "queue",
+            &input.name,
+            input.rate_limit.as_deref(),
+        )?,
         max_concurrent: input.max_concurrent,
     })
 }
diff --git a/sdks/node/test/worker/rateScope.test.ts b/sdks/node/test/worker/rateScope.test.ts
new file mode 100644
index 00000000..303dbc79
--- /dev/null
+++ b/sdks/node/test/worker/rateScope.test.ts
@@ -0,0 +1,22 @@
+import { mkdtempSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { expect, it } from "vitest";
+import { Queue } from "../../src/index";
+
+function newQueue(): Queue {
+  return new Queue({ dbPath: join(mkdtempSync(join(tmpdir(), "taskito-sc-")), "q.db") });
+}
+
+it("names the task for a malformed task rate limit", () => {
+  const queue = newQueue();
+  queue.task("t", () => {}, { rateLimit: "100/mm" as never });
+  expect(() => queue.runWorker({ queues: ["default"] })).toThrow(/on task 't'/);
+});
+
+it("names the queue, not the task, for a malformed queue rate limit", () => {
+  const queue = newQueue();
+  queue.task("t", () => {});
+  queue.configureQueue("default", { rateLimit: "100/mm" as never });
+  expect(() => queue.runWorker({ queues: ["default"] })).toThrow(/on queue 'default'/);
+});

From 355fab7b47b2aa29e05008ef7d27b0890c7eb4ea Mon Sep 17 00:00:00 2001
From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com>
Date: Thu, 16 Jul 2026 17:07:10 +0530
Subject: [PATCH 13/13] test(node): assert the whole rate error, not a
 substring
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

toThrow(string) only checks containment, so a message naming both scopes would
have satisfied it — and naming exactly one is the contract under test. Compare
the whole message.
---
 sdks/node/test/worker/rateScope.test.ts | 28 +++++++++++++++++++++----
 1 file changed, 24 insertions(+), 4 deletions(-)

diff --git a/sdks/node/test/worker/rateScope.test.ts b/sdks/node/test/worker/rateScope.test.ts
index 303dbc79..912ae464 100644
--- a/sdks/node/test/worker/rateScope.test.ts
+++ b/sdks/node/test/worker/rateScope.test.ts
@@ -8,15 +8,35 @@ function newQueue(): Queue {
   return new Queue({ dbPath: join(mkdtempSync(join(tmpdir(), "taskito-sc-")), "q.db") });
 }
 
-it("names the task for a malformed task rate limit", () => {
+/**
+ * The message a failed `runWorker` threw.
+ *
+ * `toThrow(string)` only checks containment, and the contract here is that
+ * exactly one scope is named — a message mentioning both would satisfy a
+ * substring match. So compare the whole message instead.
+ */
+function startError(queue: Queue): string {
+  try {
+    queue.runWorker({ queues: ["default"] });
+  } catch (error) {
+    return (error as Error).message;
+  }
+  throw new Error("runWorker should have rejected the malformed rate limit");
+}
+
+// Tasks and queues configure a rate limit through one parser, so the error has
+// to name which kind is wrong.
+it("names the task, and only the task, for a malformed task rate limit", () => {
   const queue = newQueue();
   queue.task("t", () => {}, { rateLimit: "100/mm" as never });
-  expect(() => queue.runWorker({ queues: ["default"] })).toThrow(/on task 't'/);
+  expect(startError(queue)).toBe("invalid rateLimit '100/mm' on task 't' (expected e.g. '100/m')");
 });
 
-it("names the queue, not the task, for a malformed queue rate limit", () => {
+it("names the queue, and only the queue, for a malformed queue rate limit", () => {
   const queue = newQueue();
   queue.task("t", () => {});
   queue.configureQueue("default", { rateLimit: "100/mm" as never });
-  expect(() => queue.runWorker({ queues: ["default"] })).toThrow(/on queue 'default'/);
+  expect(startError(queue)).toBe(
+    "invalid rateLimit '100/mm' on queue 'default' (expected e.g. '100/m')",
+  );
 });