Workers

run_worker() blocks the calling thread until the worker stops — that's why the CLI (flexiq worker --app ...) is the recommended way to run one. To keep your own thread free, run it in a background thread or use the async variant:

import threading

t = threading.Thread(target=queue.run_worker, daemon=True)
t.start()

import asyncio

async def main():
    await queue.arun_worker()  # runs the worker loop in a thread executor

asyncio.run(main())

runWorker() never blocks — it starts the worker on the Rust core and returns immediately. The bundled flexiq run ./app.js CLI wraps this for you and wires Ctrl+C to worker.stop(); calling runWorker() from your own code needs no CLI at all.

start() never blocks — scheduling runs in the background and handlers execute on the pool you configure. There's no CLI worker subcommand: workers are code, built with flexiq.worker() in your application; wire awaitShutdown() (see Graceful shutdown) to keep the process alive.

spawn() never blocks: the scheduler, the handler pool, the result drain and the heartbeat each get their own thread, and the call returns a WorkerHandle as soon as they are up. Nothing keeps main alive for you, so a worker process has to park itself — block on a channel your signal handler writes to, then call shutdown():

let worker = queue.worker().register::<process>().spawn()?;

// Park until something asks this process to stop. `rx` is yours — a channel a
// signal handler writes to, an admin endpoint, a shutdown flag.
let _ = rx.recv();

worker.shutdown()?;

Dropping the handle instead leaves the worker running detached — the threads are still there, still claiming jobs, with nothing left to stop them. shutdown() is the only way back.

The worker registers itself as sdk = "rust" with a pool_type of "rust", and takes the id rust-worker-<uuid7> unless .worker_id(..) names one. worker.worker_id() reads it back.

@queue.task() registers a task on the Queue object the moment its module is imported. run_worker() serves whatever's registered on self at call time — make sure task modules are imported (directly, or via the --app path) before the worker starts.

Tasks declared with the queue-less @flexiq.task() are claimed automatically at Queue(...) construction, at queue.autodiscover(...), and again when run_worker() starts — so a worker entrypoint that imports its task modules needs no extra call. See Task discovery.

queue.task(name, handler) registers a task on the Queue instance immediately. runWorker() serves whatever's registered on that instance, so import your task modules before calling it.

Tasks declared with the queue-less task(name, handler) are claimed automatically when the Queue is constructed, by await queue.discover(...), and again when runWorker() or runExecutor() starts — so a worker entrypoint that imports its task modules needs no extra call. See Task discovery.

There's no global registry a worker drains — each worker handles what it binds with .handle(task, fn) (or .register(...) / .apply(...) for generated handler sets), or what .discover() loads from the @TaskHandler companions the annotation processor listed in META-INF/services. A job for a task name the builder never bound fails immediately with no handler registered for task '...', which also means two workers can bind different subsets of your tasks from the same FlexiQ client.

Discovery is explicit and never replaces a handler already on the builder — see Task discovery.

There is no global registry a worker drains. Each worker runs exactly the tasks in its own register::<T>() chain, so two workers over the same FlexiQ can serve different subsets on purpose:

let billing = queue.worker().register::<charge>().queues(["billing"]).spawn()?;
let general = queue.worker().register::<resize>().register::<total>().spawn()?;

register::<T>() takes the type #[flexiq::task] generated, so a task that was renamed or deleted breaks the build rather than dead-lettering at runtime. What still compiles is a task nobody added to the chain: the first job for it fails fatally with task not registered on this worker, because retrying cannot make it runnable here. Registering one name twice is refused by spawn().

Nothing about this is implicit, so there is nothing to discover — see Task discovery.

Advertise capability tags when starting a worker:

queue.run_worker(tags=["gpu", "heavy"])

Tags are stored with the worker's registration and show up in queue.workers() and the dashboard — a lightweight way to see which machines have which capabilities at a glance. They're informational: the scheduler doesn't filter dispatch by tag, so route GPU-only or heavy-workload jobs to the right machines with dedicated queues instead.

Node workers can't advertise capability tags — runWorker() has no tags option. Route capability-specific work with dedicated queues instead: give GPU or heavy jobs their own queue and start a worker with runWorker({ queues: ["gpu"] }).

listWorkers() rows include a tags field, but only SDKs that can set it populate it — it stays empty for Node-started workers.

flexiq.worker() has no tags(...) builder method — Java workers can't advertise capability tags. Route capability-specific work with dedicated queues instead (flexiq.worker().queues("gpu").start()).

WorkerInfo.tags exists on listWorkers() rows, but only SDKs that support tag advertisement set it; it's null for Java-started workers.

WorkerBuilder has no tags(..), so a Rust worker advertises none. The registration record has a tags field and this shell leaves it unset, which is why list_workers() rows for a Rust worker carry None there.

Route capability-specific work with dedicated queues instead — give the GPU or heavy jobs their own queue and start a worker on it:

queue.worker().register::<render>().queues(["gpu"]).spawn()?;

The dispatch that tags would inform does not exist anyway: the scheduler routes by queue and task name, never by tag.

workers sets the OS-thread pool size (threads share a single GIL — only one runs Python bytecode at a time). Swap in the prefork pool for true CPU parallelism, or tune async_concurrency for async def tasks, which run on a dedicated event loop instead of the thread pool.

There's no worker-level thread count — handlers run concurrently on the single event loop, each an independent async invocation. concurrency caps how many attempts run at once on this worker; unset, it is unbounded, and the worker claims jobs faster than it runs them, stranding them running where other workers cannot take them. It counts attempts, not live handlers — a timed-out attempt frees its slot while its handler may still be running. batchSize controls how many jobs are claimed per poll, and per-task/queue maxConcurrent caps a task or queue across every worker. See Execution model.

Jobs run on real JVM threads — no GIL, no event loop to block. concurrency(n) fixes the handler pool size; autoscale(...) resizes it with queue depth instead. See Execution model for CPU-bound sizing guidance.

Jobs run on real OS threads — every handler is synchronous and runs on a blocking thread, so num_workers is both the pool size and the concurrency cap. A value below 1 is raised to 1.

It also becomes the scheduler's max_in_flight when you do not set one, so the worker never claims more jobs than it can run. Override that deliberately with .scheduler_config(SchedulerConfig { max_in_flight: Some(n), ..Default::default() }).

#[flexiq::task] refuses an async fn outright rather than quietly blocking a pool thread. If a task really has to be a future, flexiq_core::Worker exposes register_async, which is core's own pool and not this shell's — so it does not carry the durable-step fence. Otherwise, drive the runtime inside the body.

The heartbeat runs on a background thread every 5 seconds and carries the current resource health snapshot, which is how queue.resource_status() sees health from a separate dashboard process.

Lifecycle events

Subscribe to worker join/leave/health transitions:

from flexiq import EventType

@queue.on_event(EventType.WORKER_ONLINE)
def on_online(event_type, payload):
    print(f"Worker {payload['worker_id']} joined")

@queue.on_event(EventType.WORKER_OFFLINE)
def on_offline(event_type, payload):
    print(f"Worker {payload['worker_id']} went away")

The heartbeat fires every 5 seconds and carries the current resource-health snapshot, so worker resources going unhealthy shows up in listWorkers() too.

Lifecycle events

Node has no worker-lifecycle events. queue.on(event, handler) fires only the four job-outcome events (job.completed, job.retrying, job.dead, job.cancelled) — there's no WORKER_ONLINE / OFFLINE / UNHEALTHY. Observe worker presence and health by polling listWorkers(), whose rows carry status and a per-resource resourceHealth snapshot from the last heartbeat.

Heartbeating is managed natively by the core — there's no user-facing interval to configure.

Lifecycle events

Java has no worker-lifecycle events. .on(EventName, ...) fires only the four job-outcome events (SUCCESS, RETRY, DEAD, CANCELLED) — there's no WORKER_ONLINE / OFFLINE / UNHEALTHY. Observe worker presence and health by polling flexiq.listWorkers(), whose WorkerInfo rows carry status and a resourceHealth snapshot from the last heartbeat.

spawn() starts a dedicated heartbeat thread that ticks every 5 seconds. There is no interval to configure. The same tick does two cluster chores: the leader-elected reap of workers that stopped heartbeating, and the sweep of ephemeral subscriptions whose owner is gone.

Lifecycle events

The Rust SDK has no event bus, so there is no WORKER_ONLINE / OFFLINE / UNHEALTHY to subscribe to. Observe presence by polling queue.storage().list_workers(), whose rows carry status and the last heartbeat.

For job outcomes rather than worker ones, the nearest seam is flexiq_core::Worker::on_outcome — a callback handed every processed success, retry, dead-letter and cancel. WorkerBuilder does not expose it, and building the worker through flexiq_core::Worker directly costs the durable-step fence this shell's pool carries.

run_worker() wires Ctrl+C (SIGINT/SIGTERM) for you: the first signal starts a warm shutdown — stop claiming new jobs, wait up to drain_timeout seconds for in-flight ones — and a second signal force-kills immediately.

queue = Queue(db_path="myapp.db", drain_timeout=60)  # wait up to 60s (default: 30)

$ flexiq worker --app myapp:queue
[flexiq] Starting worker...
^C
[flexiq] Warm shutdown (waiting for running tasks to finish)...
[flexiq] Worker stopped.

runWorker() doesn't wire any signal handling itself — the flexiq run ./app.js CLI does that for you (SIGINT/SIGTERM → worker.stop()). Calling runWorker() from your own code, wire it yourself:

process.on("SIGTERM", () => worker.stop());

There's no automatic signal handling either — wire a shutdown hook so the JVM calls close() on exit, and block main on awaitShutdown():

Runtime.getRuntime().addShutdownHook(new Thread(() -> {
    worker.close();
    flexiq.close();
}));
worker.awaitShutdown();

Nothing wires SIGINT or SIGTERM for you, and there is no drain timeout to configure: shutdown() stops dispatch, waits for every in-flight handler, stops the heartbeat, joins all three threads and unregisters the worker. It blocks for as long as the slowest running handler takes.

It also takes the handle by value, so a worker cannot be stopped twice — which means the handle has to reach whatever triggers the stop. Put it behind an Arc<Mutex<Option<WorkerHandle>>>, or keep the handle on the main thread and have the signal path only wake it.