FyFlow Scheduler

Zero-dependency parallel task scheduler with resource management and cross-platform support.

Using this library from an AI agent? AGENTS.md is a complete single-file reference: exact signatures, defaults, events, gotchas and recipes. Every snippet in it is executed by the test suite.

Overview

FyFlow is a self-contained parallel task scheduler with resource management. Fully in-memory operation with zero dependencies.

Installation

Node.js and browser - npm, which ships prebuilt bundles with the worker bootstrap inlined, so no bundler configuration is needed:

npm install fyflow-scheduler
# or
yarn add fyflow-scheduler

Deno - JSR, which ships the TypeScript sources and loads the worker directly:

deno add jsr:@fyflow/scheduler

The two packages are not interchangeable. The JSR package is Deno-only: it contains no Node-specific files, and importing it from Node or a browser throws an error pointing you at npm. Node and browser support lives entirely in the npm package.

Quick Start

import { FyflowScheduler, FyflowTask, WorkerManager, ConcurrentLimitGroup } from 'fyflow-scheduler';

// A pool must DECLARE the groups it uses - registering a group on the scheduler
// alone does not constrain anything
const workerPool = new WorkerManager('./worker.js', { maxThreads: 2, groups: ['cpu'] });

const scheduler = new FyflowScheduler(
  { MyWorker: workerPool },
  { cpu: new ConcurrentLimitGroup(4) }
);

const tasks = [
  new FyflowTask({ id: 'task1', workerType: 'MyWorker', payload: { data: 'input' } }),
  new FyflowTask({ id: 'task2', workerType: 'MyWorker', payload: { data: 'processed' } })
];

// createPromise is required to get results back - addTask is fire-and-forget
const results = await Promise.all(scheduler.addTasks(tasks, { createPromise: true }));

await scheduler.shutdown();   // required, or live workers keep the process alive

See Loading Workers for how ./worker.js is resolved - it differs between Deno, Node and the browser.

Key Features

• Parallel Task Execution: High-performance concurrent task processing
• Resource Management: Named concurrency and rate-limit groups applied per pool or per task
• Resident Resources: Groups held for a worker's lifetime, with per-worker costs - for a model on a GPU, a connection, a licence seat
• Dynamic Task Spawning: Workers can create new tasks at runtime
• Cross-Platform: Node.js, Browser, and Deno support
• Worker Pools: Thread-based and inline execution modes
• Progress Reporting: Real-time task progress and event monitoring
• Worker Messages: Workers emit() named events - logs, metrics, partial results - to the host, batched, bounded and sent only while someone listens
• Cancellation: Stop a waiting or running task; workers see an AbortSignal, and a threaded worker that ignores it is stopped for them
• Zero Dependencies: Self-contained with no external requirements
• In-Memory: Fast coordination, with nothing persisted

Basic Usage

Creating Workers

A worker script must export default a class.

import { BaseWorker } from 'fyflow-scheduler';

export default class MyWorker extends BaseWorker {
  // The pool calls `new MyWorker(config, workerContext)` - forward both
  constructor(config = {}, workerContext) {
    super(config, workerContext);
  }

  // setup and teardown are abstract on BaseWorker: both required, may be empty
  async setup() {}
  async teardown() {}

  async run(payload, context) {
    context?.sendProgress(0.5, "Processing...");   // progress is 0-1

    // Spawn additional tasks if needed
    context?.spawnTask({
      id: 'child-task',
      workerType: 'MyWorker',
      payload: { data: 'child-data' }
    });

    return { result: 'completed', data: payload.data };
  }
}

Loading Workers

Point the pool at your worker file's URL.

npm (Node and browser) - a compiled .js worker always works:

const workerUrl = new URL('./myWorker.js', import.meta.url).href;
const pool = new WorkerManager(workerUrl, { maxThreads: 4 });

No bundler plugin and no build step for the worker itself. On Node >= 22.18 a .ts worker works too, with no build step, as long as it sticks to erasable syntax - types must be imported with import type, and enum is out. See AGENTS.md §3 for the full conditions. In the browser there is no type stripping, so the worker must be JavaScript by the time it reaches the page, and your bundler needs to emit it as a separate asset - which most do automatically for new URL('./myWorker.js', import.meta.url).

Deno (JSR) - point at the TypeScript source, which Deno loads directly:

const workerUrl = new URL('./myWorker.ts', import.meta.url).href;

The ?worker-direct suffix in this repository's examples and tests is a convention of its own esbuild config, used so the .ts examples run on both runtimes. It is not part of the published API and will not resolve in your code.

Getting Results Back

addTask and addTasks are fire-and-forget by default and return nothing. Ask for a promise explicitly:

const result = await scheduler.addTask(task, { createPromise: true });

const results = await Promise.all(
  scheduler.addTasks(tasks, { createPromise: true })
);

await scheduler.shutdown();   // required, or workers keep the process alive

Parallel Task Execution

const tasks = [
  new FyflowTask({ id: 'fetch', workerType: 'DataWorker', payload: { url: 'api/data' } }),
  new FyflowTask({ id: 'validate', workerType: 'ValidationWorker', payload: { url: 'api/data2' } }),
  new FyflowTask({ id: 'process', workerType: 'ProcessWorker', payload: { batch: 1 } }),
  new FyflowTask({ id: 'save', workerType: 'SaveWorker', payload: { batch: 2 } })
];

// All tasks execute in parallel (subject to resource constraints)
tasks.forEach(task => scheduler.addTask(task));

Resource Groups

const cpuGroup = new ConcurrentLimitGroup(8);  // Max 8 concurrent CPU tasks
const gpuGroup = new ConcurrentLimitGroup(2);  // Max 2 concurrent GPU tasks

const scheduler = new FyflowScheduler(workerPools, {
  cpu: cpuGroup,
  gpu: gpuGroup
});

// Workers automatically use groups specified in WorkerManager
const gpuWorkerPool = new WorkerManager('./gpu-worker.js', {
  maxThreads: 2,
  groups: ['gpu']  // This pool uses GPU constraints
});

Resident Groups

groups is scoped to a task - taken at dispatch, released when the task settles. That cannot bound a resource the worker holds simply by existing, like a model loaded in setup(). The worker outlives its task by the idle timeout, so a second pool can load a second model onto a full GPU even with a ConcurrentLimitGroup(1) behaving perfectly correctly.

residentGroups is held from worker creation to teardown, with a cost per worker:

const vram = new ConcurrentLimitGroup(24, 'vram');   // 24GB of VRAM

// A 20GB model - only one such worker fits
const big = new WorkerManager('./llama-worker.js', {
  maxThreads: 1,
  residentGroups: { vram: 20 },
  idleTimeout: 2000            // how long it keeps the GPU after its last task
});

// 2GB models - several coexist, but not alongside the big one
const embed = new WorkerManager('./embed-worker.js', {
  maxThreads: 4,
  residentGroups: { vram: 2 }
});

// Array form is shorthand for a cost of 1 per worker
const one = new WorkerManager('./worker.js', { residentGroups: ['gpu'] });

const scheduler = new FyflowScheduler({ Big: big, Embed: embed }, { vram });

A worker is not created while its cost does not fit, and tasks needing one wait until a holder is torn down. Release happens on the ordinary idle-timeout path, so a pool with work still arriving keeps its worker and its loaded model - avoiding a reload is usually worth more than the wait. Admission is head-of-line, so an expensive pool is not starved by cheaper ones queued behind it.

Costs must be positive integers no larger than the group's limit, and the group must exist; all three are rejected at construction.

idleTimeout: 0 never releases. That is fine when nothing else needs the group, but under contention it is a permanent deadlock - the waiting task stays pending forever. scheduler.completed correctly does not fire, but stats shows queued=0 running=0 because blocked tasks leave the queued count, so a stalled scheduler looks idle. resource.blocked names the contended group as the task is queued, and getAdmissionQueue() says what is waiting and for how many units. Give any pool sharing a contended resident group a non-zero idleTimeout.

See examples/resident-groups.ts for a runnable walkthrough.

Worker Messages

A worker sends named events to the host with emit() - a log line, a metric, a partial result. The scheduler defines no names or formats, so a log is one name in a shape you choose:

// worker, inside run()
context?.emit('log', { level: 'info', message: `fetched ${rows.length} rows` });

// host - subscribe before adding tasks
scheduler.addEventListener('task.message', (e) => {
  const { taskId, attemptNumber, name, data, seq, dropped } = e.detail;
  if (name !== 'log') return;
  if (dropped) appendLog(taskId, { level: 'warn', message: `${dropped} lines dropped` });
  appendLog(taskId, { ...data, attemptNumber, seq });
});
// Each attempt's log is complete by the time its task.completed,
// task.failed or task.retrying fires

Nothing is sent while nobody listens, so emitting is free until someone subscribes. Events arrive in order per attempt, data is copied with structuredClone on both pool types, and delivery is bounded by the pool's messages option: batched every 20 ms, at most 100 per batch by default, with the excess counted in dropped. A worker with no task at hand - loading a model in setup() - emits on this.workerContext instead, which arrives as worker.message on the pool. See AGENTS.md §6 for every guarantee.

Retries and Attempts

new FyflowTask({
  id: 'flaky', workerType: 'MyWorker', payload: {},
  retryPolicy: { maxRetries: 2, backoffMs: 1000 },
  attemptsSpent: 0          // attempts already made, e.g. before a restart
});

A retry starts no earlier than backoffMs after the attempt failed, however busy the scheduler is, and task.retrying fires when one is decided. task.attemptNumber is the 1-based number of the current attempt; attempts counts retries used. A task re-added after a process restart keeps its budget if you pass what it had spent as attemptsSpent - record the highest attemptNumber seen on task.running. A requeue after a worker dies is not an attempt. See AGENTS.md §5 and the resume recipe in §10.

Cancelling Tasks

import { CancelledError } from 'fyflow-scheduler';

const promise = scheduler.addTask(task, { createPromise: true });

// true once the task has stopped; false if it had already finished
const cancelled = await scheduler.cancel(task.id, { reason: 'stopped by the user' });

try {
  await promise;
} catch (error) {
  if (error instanceof CancelledError) { /* error.taskId, error.reason */ }
}

A task still waiting - queued, blocked on a resource group, or in a retry backoff - is removed and never runs. A running one has its context.signal aborted, which the worker passes on or checks:

async run(payload, context) {
  const response = await fetch(payload.url, { signal: context?.signal });
  return await response.json();
}

Either way it ends cancelled, whatever its run produced, and is never retried or requeued. A threaded worker that has not stopped within the pool's cancelGraceMs (default 2000) is terminated, and its other tasks are requeued; an inline run cannot be interrupted, and is waited for. cancel() never throws for a task's state, so cancelling a whole run is a loop. See AGENTS.md §5 and §8.

Event Monitoring

scheduler.addEventListener('task.completed', (e) => {
  console.log(`Task ${e.detail.id} completed:`, e.detail.result);
});

scheduler.addEventListener('task.progress', (e) => {
  // The detail is the task's fields spread flat, so the id is `id` - there is
  // no `taskId`. (The pool's own task.progress does carry `taskId`.)
  console.log(`Task ${e.detail.id}: ${(e.detail.progress * 100).toFixed(0)}%`);
});

scheduler.addEventListener('scheduler.completed', (e) => {
  // Nothing left queued, blocked or running - whatever each task's outcome
  console.log(`Drained. Stats:`, e.detail);
});

API Reference

Core Classes

FyflowScheduler(workerPools, resourceGroups, options?)

• options.maxCompletedTasks: cap how many terminal tasks stay in scheduler.tasks (default: unlimited). Set it for long-lived schedulers - completed tasks otherwise pin their payloads and results forever
• options.periodicRetryIntervalMs: heartbeat for waking blocked tasks (default 50ms). A fallback: releases wake their own queue directly, rate windows wake theirs when they roll over, and a group emitting slot-released is heard. This covers a custom group that changes without saying so
• addTask(task, options?): Add a task. Fire-and-forget by default; pass { createPromise: true } to get a promise back
• addTasks(tasks, options?): Batch version, optimised for bulk additions. Throws on an unknown workerType, validating the batch before queueing any of it
• cancel(taskId, options?): Stop a task, waiting or running. Resolves true once it is cancelled, false if it had already finished. options.reason, options.graceMs for a threaded worker
• stats: { queued, running, done, failed, cancelled }, counting tasks rather than attempts
• getResourceMetrics() / getResourceStats(): Per-group utilisation and lifetime counters
• describeResources(): The gauges every group presents. Schema, not state - safe to read once and cache
• getAdmissionQueue(groupId?): Everything currently waiting on a resource group, in admission order, with the cost each waiter needs - the head is what explains a stall
• shutdown(): Terminate worker pools and release all listeners
• addEventListener(event, handler): Event monitoring

FyflowTask(config)

• id: Unique task identifier
• workerType: Worker pool to use
• payload: Data passed to worker
• workerGroups: Optional resource group names
• optional: Mark task as optional (failures don't block workflow)
• limitKey: Bucket for keyed resource groups (see KeyedRateLimitGroup and KeyedConcurrentLimitGroup)
• units: Requests for particular units of a UnitGroup, per group id
• retryPolicy: { maxRetries, backoffMs }
• attemptsSpent: Attempts already made, so a task re-added after a restart keeps its retry budget
• handleRejection: Silence an unhandled rejection for fire-and-forget use (default true)
• Readable as it runs: state, result, error, attempts, attemptNumber, cancellation, startTime, endTime, executionTime
• onCompletePromise(): Promise for this task alone
• onCompleteDescendants(): Promise for this task plus everything it spawns

WorkerManager(scriptUrl, options)

• maxThreads: Maximum worker instances (not OS threads). With inline: false each is a real worker thread; with inline: true they are objects in the main process, so only maxThreads × maxConcurrentTasks matters for throughput — the split controls how much per-instance state (connections, caches) is duplicated
• maxConcurrentTasks: Tasks per instance
• groups: Resource group names, acquired per task
• residentGroups: Groups held for a worker's lifetime, string[] (cost 1 each) or Record<string, number> for weighted costs. Per worker, not per pool
• inline: Use inline execution (default: false)
• config: First constructor argument for every worker instance in the pool
• requeueFailedTasks: Requeue a dead worker's tasks rather than failing them (default true)
• maxWorkerRestarts: Restarts before the pool gives up and emits worker.restart_limit_exceeded (default 3)
• messages: Bounds on emit() (default { flushIntervalMs: 20, maxBuffered: 100 })
• idleTimeout: Ms before an idle worker is terminated (0 = never, default 5000)
• idleCheckIntervalMs: How often idle workers are swept (default 5000). This is the floor on how promptly a resident group is handed over
• cancelGraceMs: Ms a threaded worker gets to stop a cancelled task before it is terminated (default 2000). A forced stop is not counted as a restart

ConcurrentLimitGroup(limit, id?)

• Resource constraint with a concurrent execution limit. A hard cap: no more tasks hold it than its limit

RateLimitGroup(windows, id?)

• Time-window throttling, e.g. [{ limit: 10, windowMs: 1000 }]. Multiple overlapping windows are enforced together

KeyedRateLimitGroup(windows, options?)

• The same windows applied independently per key, for when each endpoint, tenant or account has its own quota
• options.keyFrom derives the key from a task, defaulting to task.limitKey
• A task with no derivable key throws at addTask rather than silently sharing a bucket
• Blocked tasks queue per key, so a saturated key never delays another
• Idle keys are evicted after options.idleKeyTtlMs (default: twice the largest window), so high-cardinality keys do not grow without bound
• getKeyMetrics(key) and getActiveKeys() give the per-bucket view

KeyedConcurrentLimitGroup(limit, options?)

• A concurrent limit per key - one disk, one database each - so a saturated key never delays another. Keyed like KeyedRateLimitGroup
• options.limitFor(key) gives a key its own limit, e.g. 1 for a slow disk
• keyFrom may return several keys - a C: -> D: copy holds both disks, and starts only when both have room
• A key is dropped as soon as nothing holds it

UnitGroup(units, options?)

• A limit whose units have names - two GPUs, licence seats - and the work is told which it got: context.grants.gpu is ['gpu1'] in run()
• In residentGroups it gives one worker per device: each worker holds its units for its lifetime and reads them from this.workerContext.grants, from its constructor on
• A task or a pool can ask for particular units: units: { gpu: { from: ['gpu1'] } } on a task, residentGroups: { gpu: { from: ['gpu0', 'gpu1'] } } on a pool, and prefer as a hint that never blocks. A pinned waiter holds up nobody, and a freed unit goes to the oldest waiter that can use it
• new UnitGroup(2) names its units 0 and 1. A custom group can grant too: set grants: true and return the grant from onStart - see AGENTS.md §5

Worker management (on a WorkerManager)

• getWorkerIds(): ids of the workers that currently exist (created lazily)
• getWorkerStatus(id) / getAllWorkerStatuses(): { id, state, tasksCompleted, errorCount, uptime, currentTasks, resourcesHeld, lastError? }
• restartWorker(id, newConfig?) / replaceWorker(...): terminate and rebuild a worker, optionally with new config. The replacement gets a new id
• updateWorkerConfig(id, config): merge config into a live worker
• getResidentUsage(): units this pool currently holds in each resident group
• getResidentGrants(id): what a worker's resident groups granted it
• shutdown(): terminate the pool

Worker self-termination — a worker can ask to be torn down, e.g. after detecting a bad connection. It is a request, never a throw, so it is safe from a timer. Called on the worker's context, every in-flight task is requeued when the pool's requeueFailedTasks allows it:

this.workerContext?.terminateWithError(new Error('connection lost'), { canRestart: true });

Called on a task's context instead, that task fails with a WorkerTerminationError and only the others are requeued. See AGENTS.md §4.

When Workers Fail

The scheduler never discards queued work to make a broken pool look healthy:

• A worker that dies or cannot be constructed, with requeueFailedTasks: true (the default), has its tasks requeued. The pool retries up to maxWorkerRestarts times, then gives up. If it never manages to run a worker those tasks wait, and their promises stay pending - a pool that cannot keep or build a worker needs fixing, not its work thrown away.
• With requeueFailedTasks: false the tasks fail and their promises reject.

So watch the pool, not only the task. worker.restart_limit_exceeded means the pool has given up rebuilding workers and needs intervention - it covers both a worker that cannot be built and one that keeps dying.

Events

On the scheduler: task.running, task.retrying, task.completed, task.failed, task.cancelling, task.cancelled, task.progress, task.message, task.user_action, task.spawn_request, task.spawn_failed, scheduler.completed, resource.acquired, resource.released, resource.blocked, resource.unblocked. Every one carries a timestamp.

The four resource.* events let a resource view be a fold of a stream rather than a poll of four accessors that describe four different instants. They fire on the scheduler only - a WorkerManager never dispatches them, so a consumer subscribing to both cannot double-count. See AGENTS.md section 6 for the payload, the five ways a blocked task leaves its queue, and the two conservation invariants worth folding against.

On a WorkerManager: task.started, task.completed, task.failed, task.progress, task.spawn_request, task.requeue_required, worker.failed, worker.self_terminated, worker.restart_limit_exceeded, worker.message, plus the worker lifecycle events worker.initialization.started|completed|failed, worker.setup.started|completed and worker.teardown.started|completed|failed.

An event detail is a live object, not a snapshot. For task.running, task.retrying, task.completed, task.failed, task.user_action, task.cancelling and task.cancelled it is the FyflowTask the scheduler keeps mutating, so arr.push(e.detail) gives you entries that all read the terminal state later. Listeners run synchronously - project the fields you need inside the listener instead. task.progress, task.message, worker.message, scheduler.completed and the resource.* events are unaffected; their details are copies.

Lifecycle events carry { workerId, workerType, timestamp }, plus duration on *.completed and error on *.failed. Inline and threaded pools emit the same set. They fire on every worker creation and idle-timeout teardown, so keep those listeners cheap.

scheduler.completed fires whenever the scheduler drains - whether its tasks completed, failed or were cancelled - so it can fire more than once if tasks arrive in waves.

Worker Interface

Workers extend BaseWorker and implement:

• constructor(config): Setup worker configuration
• async run(payload, context): Execute task logic. context carries sendProgress(), emit(), spawnTask(), terminateWithError() and signal, the AbortSignal that cancel() aborts
• async setup(): Initialize worker (required by BaseWorker; may be empty)
• async teardown(): Cleanup worker (required by BaseWorker; may be empty)

Performance

Measured with npm run benchmark:quick:

• Throughput (inline): tens of thousands of tasks/sec, roughly flat from 10K to 100K tasks - queueing and cancelling cost the same per task at any queue length
• Efficiency (threaded, CPU work): 96-98%. This is achieved average concurrency over configured concurrency - "how much of the parallelism I asked for did I get" - so steady-state coordination overhead is a few percent
• Memory: completed tasks are retained by default, with their payloads and results, so a long-running scheduler grows until maxCompletedTasks is set

Worker startup is measured separately (npm run benchmark:startup), because every other scenario pre-warms its pool. It is the largest difference between the two worker types, and between runtimes:

| | Deno | Node 22 | |---|---:|---:| | 1 worker thread | ~12 ms | ~54 ms | | 8 inline instances (each) | ~85 µs | ~324 µs |

Threads cost roughly three orders of magnitude more to start than inline instances, so a short-lived threaded pool can spend most of its life starting up.

Numbers vary substantially between runs and machines, and a scenario's result depends on what ran before it in the suite. Re-run the suite rather than relying on these figures, and compare full-suite runs only against other full-suite runs.

Platform Support

| Platform | Package | Worker Threads | Inline Workers | Build Required | |----------|---------|---------------|----------------|----------------| | Node.js | npm fyflow-scheduler | ✅ | ✅ | Prebuilt in the package | | Browser | npm fyflow-scheduler | ✅ | ✅ | Prebuilt in the package | | Deno | JSR @fyflow/scheduler | ✅ | ✅ | None - sources are loaded directly |

Only four files differ between runtimes, and the split is resolved at build time rather than at runtime: the worker bootstrap (workerWrapper.ts vs workerWrapper.node.ts), the URL that locates it, and two feature checks in ThreadWrapper for Node's worker_threads event emitter. Everything else - the scheduler, worker manager and all resource groups - is platform-agnostic.

Node.js 22 or newer is required: the library and its tests rely on CustomEvent being a global, which landed in Node 19.

Development

# Build library
npm run build

# Run tests - all nine suites, the documentation examples among them
npm test                    # Node.js tests (requires Node >= 22)
npm run test:browser        # Browser tests (Playwright)
npm run test:deno           # Deno tests
npm run test:docs           # Just the executable documentation examples
npm run test:performance    # Contention scaling (not part of the default run)

# Type check library, tests, examples and benchmarks
deno task check

# Benchmarks
npm run benchmark           # Full benchmark suite
npm run benchmark:quick     # Quick performance test
npm run benchmark:startup   # Worker startup cost, for comparing runtimes
npm run benchmark:baseline  # The comprehensive regression set

Examples

See the examples/ directory for comprehensive usage examples:

• getting-started.ts: Basic parallel task execution
• enhanced-features.ts: Progress reporting, dynamic task spawning and onCompleteDescendants()
• worker-types.ts: Thread vs inline worker comparison
• performance-groups.ts: Resource management, rate limits and constraints
• resident-groups.ts: Worker-lifetime resources and weighted per-worker costs
• monitor-resource-health.ts: Drop-in resource health monitor helper
• deno-only/minimal.ts: The plain Deno form - a worker URL with no build step and none of this repo's bundler convention

The examples above the deno-only/ folder run on both Deno and Node, which is why they carry the ?worker-direct branch. Run them all with npm run build:dev && deno task examples.

License

MIT License.

Links

• GitHub Repository
• NPM Package
• Issue Tracker