From 22de407c73dbaae56042e548c27d794b602dae53 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Wed, 15 Jul 2026 21:46:22 +0530 Subject: [PATCH 1/2] fix(node): bound in-flight jobs to channelCapacity 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 is never full applies no backpressure, and max_in_flight was never set, so nothing capped concurrency: the worker claimed jobs it could not run, stranding the surplus Running and starving peers on the same database. A 12-job backlog ran 12-wide against a documented bound of 128. Take a permit before spawning and set max_in_flight to the same number, so channelCapacity means what its docs already claim. --- crates/taskito-node/src/dispatcher.rs | 27 ++++++++-- crates/taskito-node/src/worker.rs | 7 ++- sdks/node/test/worker/backpressure.test.ts | 61 ++++++++++++++++++++++ 3 files changed, 89 insertions(+), 6 deletions(-) create mode 100644 sdks/node/test/worker/backpressure.test.ts diff --git a/crates/taskito-node/src/dispatcher.rs b/crates/taskito-node/src/dispatcher.rs index 42f8b2c1..124dbb6d 100644 --- a/crates/taskito-node/src/dispatcher.rs +++ b/crates/taskito-node/src/dispatcher.rs @@ -5,6 +5,7 @@ //! `Promise`, 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; 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, } 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, ) { // 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..ad57c5d6 100644 --- a/crates/taskito-node/src/worker.rs +++ b/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; }); diff --git a/sdks/node/test/worker/backpressure.test.ts b/sdks/node/test/worker/backpressure.test.ts new file mode 100644 index 00000000..d2614303 --- /dev/null +++ b/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 { + 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); +}); From a9a4633fc322b8544f1b76f6cef24662f06e6eeb Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Wed, 15 Jul 2026 21:46:35 +0530 Subject: [PATCH 2/2] feat(node): surface maxInFlightPerTask The per-task bulkhead already lives in the core scheduler; this exposes it on the Node task options, so one slow task cannot take a whole worker's slots. Java keeps it unset: its task config carries retry policy only, and the in-process cap makes little sense without max_concurrent beside it. --- crates/taskito-java/src/worker.rs | 5 ++++- crates/taskito-node/src/config.rs | 4 ++++ crates/taskito-node/src/convert/task_config.rs | 3 +-- sdks/node/src/types.ts | 6 ++++++ sdks/node/src/worker.ts | 2 ++ 5 files changed, 17 insertions(+), 3 deletions(-) diff --git a/crates/taskito-java/src/worker.rs b/crates/taskito-java/src/worker.rs index 1b67c709..d5c80cfc 100644 --- a/crates/taskito-java/src/worker.rs +++ b/crates/taskito-java/src/worker.rs @@ -278,7 +278,10 @@ fn register_task_policies(scheduler: &mut Scheduler, configs: Option, pub retry_max_delay_ms: Option, pub max_concurrent: Option, + /// 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, /// Rate-limit spec like `"100/m"`, `"50/s"`, `"3600/h"`. pub rate_limit: Option, pub circuit_breaker: Option, diff --git a/crates/taskito-node/src/convert/task_config.rs b/crates/taskito-node/src/convert/task_config.rs index 7b22c249..996be37a 100644 --- a/crates/taskito-node/src/convert/task_config.rs +++ b/crates/taskito-node/src/convert/task_config.rs @@ -42,8 +42,7 @@ pub fn task_config(input: &TaskConfigInput) -> Result { .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), }) } diff --git a/sdks/node/src/types.ts b/sdks/node/src/types.ts index eb3e2cdc..c57408f7 100644 --- a/sdks/node/src/types.ts +++ b/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. */ diff --git a/sdks/node/src/worker.ts b/sdks/node/src/worker.ts index 7b78c165..8a45d92b 100644 --- a/sdks/node/src/worker.ts +++ b/sdks/node/src/worker.ts @@ -318,6 +318,7 @@ function buildTaskConfigs(tasks: ReadonlyMap): 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): TaskConfi retryBaseDelayMs: options.retryBackoff?.baseMs, retryMaxDelayMs: options.retryBackoff?.maxMs, maxConcurrent: options.maxConcurrent, + maxInFlightPerTask: options.maxInFlightPerTask, rateLimit: options.rateLimit, circuitBreaker: options.circuitBreaker, });