fix(node): bound in-flight jobs; surface per-task bulkhead

crates/taskito-java/src/worker.rs

@@ -278,7 +278,10 @@ fn register_task_policies(scheduler: &mut Scheduler, configs: Option<Vec<TaskRet
                 rate_limit: None,
                 circuit_breaker,
                 max_concurrent: None,
-                // Not surfaced on this SDK yet; the whole pool stays available.
+                // Neither concurrency cap is surfaced here: this struct carries
+                // retry policy, and the in-process bulkhead only makes sense
+                // alongside max_concurrent, its cluster-wide counterpart. Both
+                // belong to a task-config surface this SDK does not have yet.
                 max_in_flight_per_task: None,
             },
         );

crates/taskito-node/src/config.rs

@@ -109,6 +109,10 @@ 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 in-flight slots, so a slow task
+    /// cannot occupy the pool and starve the others. Unlike `maxConcurrent` (the
+    /// cluster-wide cap, which costs a database read) this one is in-process.
+    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>,

crates/taskito-node/src/convert/task_config.rs

@@ -42,8 +42,7 @@ pub fn task_config(input: &TaskConfigInput) -> Result<TaskConfig> {
             .map(circuit_breaker_config)
             .transpose()?,
         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),
     })
 }
 

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,18 @@ 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.
+    concurrency: Arc<Semaphore>,
 }
 
 impl NodeDispatcher {
-    pub fn new(callback: TaskCallback, storage: StorageBackend) -> Self {
-        Self { callback, storage }
+    pub fn new(callback: TaskCallback, storage: StorageBackend, max_in_flight: usize) -> Self {
+        Self {
+            callback,
+            storage,
+            concurrency: Arc::new(Semaphore::new(max_in_flight.max(1))),
+        }
     }
 }
 
@@ -41,13 +49,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

crates/taskito-node/src/worker.rs

@@ -90,6 +90,11 @@ pub fn start_worker(
     if let Some(batch) = options.batch_size {
         config.batch_size = batch.max(1) as usize;
     }
+    // Bound in-flight work to what the dispatcher will actually run at once, so
+    // this worker never claims more than it can execute and starve peers sharing
+    // the database. `channel_capacity` is documented as the in-flight bound, and
+    // the dispatcher is gated on the same number.
+    config.max_in_flight = Some(capacity);
 
     // 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 +183,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, capacity);
     spawn(async move {
         dispatcher.run(job_rx, result_tx).await;
     });

sdks/node/src/types.ts

@@ -100,6 +100,12 @@ export interface TaskOptions {
   timeoutMs?: number;
   /** Cap on concurrently-running jobs of this task. */
   maxConcurrent?: number;
+  /**
+   * Cap on this task's share of one worker's in-flight slots, so a slow task
+   * cannot occupy the pool and starve the others. Unlike {@link maxConcurrent}
+   * (the cluster-wide cap, which costs a database read) this one is in-process.
+   */
+  maxInFlightPerTask?: number;
   /** Rate-limit spec like `"100/m"`, `"50/s"`, `"3600/h"`. */
   rateLimit?: RateLimit;
   /** Trip the task's circuit breaker after repeated failures. */

sdks/node/src/worker.ts

@@ -318,6 +318,7 @@ function buildTaskConfigs(tasks: ReadonlyMap<string, RegisteredTask>): TaskConfi
       (options.maxRetries === undefined &&
         options.retryBackoff === undefined &&
         options.maxConcurrent === undefined &&
+        options.maxInFlightPerTask === undefined &&
         options.rateLimit === undefined &&
         options.circuitBreaker === undefined)
     ) {
@@ -329,6 +330,7 @@ function buildTaskConfigs(tasks: ReadonlyMap<string, RegisteredTask>): TaskConfi
       retryBaseDelayMs: options.retryBackoff?.baseMs,
       retryMaxDelayMs: options.retryBackoff?.maxMs,
       maxConcurrent: options.maxConcurrent,
+      maxInFlightPerTask: options.maxInFlightPerTask,
       rateLimit: options.rateLimit,
       circuitBreaker: options.circuitBreaker,
     });

sdks/node/test/worker/backpressure.test.ts

@@ -0,0 +1,61 @@
+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 = 20000): 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 channelCapacity", 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.
+  //
+  // The task has to outlast several poll ticks (~50ms each) for a backlog to
+  // build: a task shorter than the poll interval finishes before the next
+  // dispatch, so the poll rate caps concurrency and the test passes either way.
+  const queue = newQueue();
+  const total = 12;
+  const capacity = 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, 600));
+    running -= 1;
+    done += 1;
+  });
+
+  for (let i = 0; i < total; i += 1) {
+    await queue.enqueue("slow");
+  }
+
+  worker = queue.runWorker({ queues: ["default"], channelCapacity: capacity });
+
+  expect(await waitFor(() => done >= total)).toBe(true);
+  expect(peak).toBeLessThanOrEqual(capacity);
+  expect(peak).toBeGreaterThan(1);
+});