Task discovery
Declare tasks without a queue, claim them in one call, and catch a worker that only discovered half of them.
Declare tasks without a queue, claim them in one call, and catch a worker that only discovered half of them.
Registering a task against a live queue or worker couples the two: the code that declares the task needs the code that built the queue, and the code that built the queue needs every task. Discovery breaks that coupling — tasks declare themselves, and a queue or worker claims them later.
The half that matters is the queue-less declaration. Walking a directory is the convenience on top; on its own it would move the error, not remove it.
Here is the shape discovery exists to undo — one place builds the queue, another declares a task against it, and neither stands alone:
# app.py
from flexiq import Queue
import tasks # the worker needs these registered before it starts
queue = Queue(db_path="app.db")# tasks.py
from app import queue # ← app.py has not reached its `queue =` line yet
@queue.task()
def send_invoice(user_id: int) -> None:
...ImportError: cannot import name 'queue' from partially initialized module 'app'
(most likely due to a circular import)Imports belong at the top of the file, which is exactly what breaks this: import tasks runs before queue is assigned, so tasks.py reaches into an app that
has not defined it yet. Moving the import below the assignment silences this
particular traceback, but the cycle is still there — and the order it depends on
is invisible to anyone reading either file on its own.
// app.ts
import { Queue } from "@byteveda/flexiq";
export const queue = new Queue({ dbPath: "app.db" });
import "./tasks/invoices.js"; // the worker needs these registered before it starts// tasks/invoices.ts
import { queue } from "../app.js"; // ← app.ts has not run its body yet
queue.task("invoices.send", (userId: string) => {
// ...
});ReferenceError: Cannot access 'queue' before initializationESM hoists imports, so tasks/invoices.ts runs its body before app.ts runs
its own — queue is still in the temporal dead zone.
// App.java — one place that has to name every handler class
Worker worker = flexiq.worker()
.apply(b -> InvoicesTasks.bind(b, new Invoices()))
.apply(b -> ReportsTasks.bind(b, new Reports()))
.apply(b -> EmailsTasks.bind(b, new Emails()))
// ...and one more line for every task class ever added
.start();Java resolves class cycles at link time, so this compiles — the cost is not an error but a central wiring list. Every new handler class means editing a file that has nothing to do with it, and a handler someone forgot to add there is invisible until a job for it fails.
// main.rs — one place that has to name every task
let worker = queue
.worker()
.register::<tasks::invoices::send_invoice>()
.register::<tasks::reports::build>()
.register::<tasks::emails::send>()
// ...and one more line for every task ever added
.spawn()?;Rust resolves the module graph at compile time, so this compiles — the cost is not an error but a central wiring list. A task left off it is invisible until a job for it lands on the worker and fails.
The half of the coupling described above does not arise: #[flexiq::task]
expands to a free type that names no queue, so a task module never imports the
module that built the handle.
The fix is a declaration that names no queue. It records the function and its options where the queue can find them later, so the task file reaches for the library and nothing else:
# tasks/invoices.py — imports flexiq, never the module holding the Queue
from flexiq import task
@task(rate_limit="10/s")
def send_invoice(user_id: int) -> None:
...@task takes exactly what
@queue.task() takes — the options
are stored as given and replayed when a queue claims the declaration, so live
objects (a predicate, middleware instances, a per-task serializer) survive
the deferral. Option names are checked against the real Queue.task signature
at decoration time, so a typo raises at import with the decorator in the
traceback.
The name follows the same rule as @queue.task(): the explicit name=, or
module.qualname.
// tasks/invoices.ts — imports the SDK, never the module holding the Queue
import { task } from "@byteveda/flexiq";
export const sendInvoice = task(
"invoices.send",
(userId: string) => {
// ...
},
{ rateLimit: "10/s" },
);task() takes exactly what
queue.task() takes — the options
are stored as given and replayed when a queue claims the declaration, so live
values (a retryOn predicate, a codec list) survive the deferral.
It returns a handle that is still the plain function: calling it runs the
handler in this process, which is what unit tests want. enqueue() on the same
handle goes through the queue that claimed it.
// Invoices.java — no Queue, no Worker, no registry
class Invoices {
@TaskHandler("invoices.send")
void send(InvoiceRequest request) {
// ...
}
}The flexiq-processor annotation processor reads @TaskHandler at compile
time and emits, per annotated class, an InvoicesTasks companion holding a
typed Task constant per handler (InvoicesTasks.SEND) plus the wiring a
worker needs. The companion also carries a nested Provider, listed as
InvoicesTasks$Provider in
META-INF/services/org.byteveda.flexiq.worker.HandlerRegistryProvider — that
entry is what makes the handlers discoverable without anything naming them.
See installation for the
Gradle and Maven setup, and Tasks for
the options @TaskHandler accepts.
A Rust declaration never names a queue in the first place — that is what
#[flexiq::task] expands to:
// tasks/invoices.rs — names the crate, and nothing that built the queue
#[flexiq::task(rate_limit = "10/s")]
pub fn send_invoice(user_id: i64) -> flexiq::Outcome<()> {
Ok(())
}What the Rust SDK does not have is the claiming half. There is no directory
walk, no module scan and no service registry: a worker runs exactly the tasks
named in its own register::<T>() chain.
let worker = queue.worker().register::<tasks::invoices::send_invoice>().spawn()?;register::<T>() takes the type the attribute generated, so a task that has
been renamed or deleted is a compile error rather than a job that dead-letters
at runtime. Registering one name twice is refused by spawn(), because one of
the two bodies would silently never run.
A declaration is inert until something claims it. How that happens differs enough per SDK to be worth reading separately.
Queue.autodiscover(package) imports the package and everything beneath it,
then claims every declaration it finds:
# app.py — imports nothing from tasks/
from flexiq import Queue
queue = Queue(db_path="app.db")
names = queue.autodiscover("tasks") # ['tasks.invoices.send_invoice', ...]The import is the load-bearing half. @task registers when its module is
executed, so walking the tree is what makes the declarations happen; the walk
alone would find nothing.
package defaults to "tasks" and may be a single module rather than a
package. The return value is the sorted names now registered on this queue — the
same list on a second call, because claiming is idempotent rather than
destructive.
You do not always need the call. A queue claims whatever is pending at four points:
| Point | What it covers |
|---|---|
Queue(...) construction | Task modules imported before the queue existed. |
autodiscover(...) | The explicit walk. |
run_worker(...) | A worker entrypoint that imported its task modules directly. |
flexiq <command> --app | An app module that imports its task modules below the queue it builds — the constructor's claim ran too early to see them. |
So a worker whose entrypoint does import myapp.tasks needs no autodiscover
at all, and no import order needs a rule. The registry is never emptied, which
is also why a second Queue in the same process receives the same tasks.
The CLI point matters most to an attached executor, which advertises the registry it reads: a name missing there is routed to nobody.
queue.discover(dir) walks a directory, imports every task module in it, then
claims every declaration it finds:
// app.ts — imports nothing from tasks/
const queue = new Queue({ dbPath: "app.db" });
const names = await queue.discover("./tasks"); // ["invoices.send", ...]The import is the load-bearing half. task() registers when its module is
executed, so walking the tree is what makes the declarations happen; the walk
alone would find nothing. It is awaited because dynamic import() is async and
ESM has no synchronous escape hatch — the one shape difference from the other
SDKs.
dir defaults to "tasks" and resolves against the working directory. The
return value is the sorted names now registered on this queue — the same list on
a second call, because claiming is idempotent rather than destructive.
The walk is depth first in name order, so registration order is the same on
every machine. It skips node_modules, anything whose name starts with a dot,
and symlinks — a symlinked directory is the cheap way to walk in a circle, and
one file reached by two paths would declare its tasks twice. It imports .js,
.mjs, .cjs, .ts, .mts and .cts, never declaration files:
// A directory holding sources next to their compiled output would import both
// x.ts and x.js — every task in them declared twice. Narrow the set instead.
await queue.discover("./dist/tasks", { extensions: [".js"] });discover() calls import() and transpiles nothing. The SDK supports Node 20,
but running a .ts file directly needs Node 22.18+ (where type stripping is on
by default) or a loader such as tsx. On an older runtime, point discover()
at your compiled output, or narrow extensions to the built files — otherwise
the first .ts module it reaches throws, and discover() surfaces that as a
TaskDiscoveryError naming the file.
You do not always need the call. A queue claims whatever is pending at four points:
| Point | What it covers |
|---|---|
new Queue(...) | Task modules imported before the queue was built — under ESM, the common case. |
discover(...) | The explicit walk. |
runWorker(...) | A worker entrypoint that imported its task modules directly. |
runExecutor(...) | The same, for an attached executor. |
So a worker whose entrypoint does import "./tasks/index.js" needs no
discover at all, and no import order needs a rule. The registry is never
emptied, which is also why a second Queue in the same process receives the
same tasks.
There is no runtime registry to drain and no directory walk. The processor
already wrote the list at compile time, so discovery is a ServiceLoader
lookup:
Worker worker = queue.worker()
.discover() // every @TaskHandler class listed in META-INF/services
.start();discover() uses the thread context class loader; discover(ClassLoader)
points it at a specific one, which is what a container or a test wants. The same
pair exists on Executor.Builder, so an
attached executor gets its
handlers the same way:
try (Executor executor = Executor.builder()
.discover()
.attach("scheduler:7777")
.start()) {
executor.awaitSession();
}The executor CLI subcommand does the same thing with no application main at
all — java -cp app.jar org.byteveda.flexiq.cli.Cli executor discovers its
handlers off the classpath and attaches.
Discovery is explicit, never implicit — queue.worker() on its own registers
nothing. A Task constant is already enqueueable the moment it is declared, so
auto-discovering inside the builder would only silently register handlers a
caller never asked for.
Nothing here scans packages or reflects over arbitrary classes. The set of handlers was fixed at compile time, which is what keeps discovery working under GraalVM native-image, where classpath scanning does not survive the closed-world build.
A provider that cannot be loaded, or that throws while building its handlers,
raises TaskDiscoveryException naming the provider class. It is never skipped:
a worker quietly missing a handler dead-letters that task's jobs instead of
failing visibly.
There is nothing to claim. A declaration is not deferred anywhere: the attribute expands at compile time into a type, and a worker names the types it wants. No import walk, no directory scan, no service file, and no queue-construction moment that quietly picks things up.
let worker = queue
.worker()
.register::<tasks::invoices::send_invoice>()
.register::<tasks::reports::build>()
.spawn()?;That chain is the registry, and writing it by hand is the whole trade. What you get for it is that the compiler checks every entry — a renamed or deleted task fails the build — and that the set is readable in one place rather than being whatever an import walk happened to reach. What you pay is the one failure discovery was invented to remove: a task nobody added to the chain compiles perfectly and is simply not served.
mod matters here, and so does cfg. A task in a module the crate never
declares is not compiled at all, and a #[cfg(..)] on the function is forwarded
onto every item the macro emits — type and both impl blocks — so a task behind
a cargo feature disappears with the feature. Naming it unconditionally then fails
the build, which is the loud outcome. Putting the chain entry behind the same
cfg is the quiet one: two builds of one binary then serve different task sets
and neither complains.
A name identifies a task across the whole system, so two declarations claiming one name is a mistake worth stopping for. Registering a task by hand replaces a name silently — that is a deliberate escape hatch. Discovery never does, because the loser would keep accepting submissions that run the winner's body.
| Situation | Behaviour |
|---|---|
| Two modules declare the same name | DuplicateTaskError at import — the traceback points at the second decorator |
A discovered name collides with a @queue.task() on the same queue | DuplicateTaskError when the queue claims it |
The same module runs twice (importlib.reload, a re-executed notebook cell) | Replaces — same declaration site, not a conflict |
@queue.task() after the queue claimed a deferred name | Replaces, silently — the escape hatch |
Pass an explicit name= to one of the two to resolve a genuine collision.
| Situation | Behaviour |
|---|---|
| Two modules declare the same name | DuplicateTaskError at import, wrapped by discover() in a TaskDiscoveryError naming the file |
A discovered name collides with a queue.task() on the same queue | DuplicateTaskError when the queue claims it |
A directory holding both x.ts and a compiled x.js | Both are imported, so the same collision — narrow extensions |
queue.task() after the queue claimed a deferred name | Replaces, silently — the escape hatch |
Give one of the two a different name to resolve a genuine collision.
| Situation | Behaviour |
|---|---|
Two @TaskHandler methods in one compilation claim one name | Compile error from the processor |
| Two providers on the classpath claim one name | DuplicateTaskException from discover(), naming both providers |
discover() would replace a handler already registered on the builder | DuplicateTaskException |
handle(...) / register(...) after discover() | Overrides, silently — the escape hatch |
Discovery never overwrites; register after it to override deliberately. Nothing is registered unless every discovered handler can be, so a clash leaves the builder as it was rather than half-discovered.
The processor's check is per compilation unit — an incremental build only sees the files it recompiled — which is why the runtime check exists as well. The two cover different halves and neither replaces the other.
| Situation | Behaviour |
|---|---|
| Two tasks in one module take the same function name | The ordinary duplicate-definition error — the macro never sees them |
| Two tasks in different modules resolve to the same task name | spawn() refuses the worker, naming the task |
| The same type registered twice in one chain | The same refusal — a register::<T>() written twice is still two |
The check is at spawn() rather than at register::<T>(), so the whole chain
reads as one expression and the error arrives with the rest of the startup
checks. It refuses rather than replacing, because one of the two bodies would
silently never run.
Note that a name collides while the types do not. Two functions named send
in two modules are distinct types and different register::<T>() arguments, but
both default their task name to "send" — which is exactly the case spawn() is
there to catch. Give one an explicit name = "..." to resolve it.
@task returns a handle that behaves like the one @queue.task() returns, so
canvases, sagas and the workflow builder accept it. Calling it runs the function
directly; .delay(), .apply_async() and .map() go through the queue that
claimed it most recently.
from tasks.invoices import send_invoice
send_invoice(42) # runs here, no queue involved
send_invoice.delay(42) # enqueues on the queue that claimed the declarationSubmitting before any queue has claimed it raises TaskNotBoundError, naming
autodiscover as the fix. Binding is last-claim-wins: with two queues in one
process, the handle submits to whichever claimed the registry most recently.
task() returns a callable handle. Calling it runs the handler directly, which
is what unit tests want; enqueue() goes through the queue that claimed it most
recently.
import { sendInvoice } from "./tasks/invoices.js";
await sendInvoice("42"); // runs here, no queue involved
sendInvoice.enqueue(["42"]); // enqueues on the queue that claimed the declarationenqueue(args?, options?) mirrors queue.enqueue(name, args?, options?), so
the two spellings stay interchangeable. It is the only part of the handle that
needs a queue: before any queue has claimed the declaration it throws
TaskNotBoundError, naming discover() as the fix. handle.bound reports
whether one has. Binding is last-claim-wins: with two queues in one process, the
handle enqueues onto whichever claimed the registry most recently.
There is no bound-versus-unbound distinction. A generated Task constant
carries its own name and payload type, so it is enqueueable the moment it is
declared — no queue has to have discovered anything first:
String id = flexiq.enqueue(InvoicesTasks.SEND, request);That is also why discovery stays on the worker side: the producer never needed it.
There is no bound-versus-unbound distinction either, and nothing to bind. The
macro leaves three things behind the function's name: send_invoice::call(..),
which builds an enqueueable value; send_invoice::run(..), the original body,
still directly callable; and the Task impl a worker registers.
send_invoice::run(42)?; // runs here, no queue involved
queue.enqueue(send_invoice::call(42))?; // goes through this queueA call is transport-free — a task name, an encoded payload and a set of options —
so it needs a queue only at enqueue, and it is the handle that names which one.
There is no last-claim-wins rule to remember, because a call is never attached to
a queue in the first place.
A reasonable-sounding alternative is to parse the files instead of importing them — read the syntax tree, find the declarations, skip running any code. It does not work here, for two reasons.
Options are runtime values. A declaration can carry a rate limit read from configuration, a predicate object, a middleware instance, a serializer. A syntax tree reports that a declaration exists; it cannot report what its arguments resolved to, and those arguments are the task's configuration.
Dispatch needs the live function. The worker calls the function itself, not a source location. Something has to load the code regardless, so parsing first would be work done twice — and the parse would be the copy free to disagree with what actually loaded.
Java gets the best of both because the annotation processor runs inside the compiler: it sees resolved types and constant annotation values, not a bare syntax tree, and it emits real code rather than a description of it. That is a compile-time registry, not a scan — and unlike reflection over the classpath, it survives a closed-world native-image build.
#[flexiq::task] runs inside the compiler as well, which answers the first
objection from the other end: it reads the function's real signature and the
attribute's literal values, so a duration, a rate or an unknown key is rejected
while it expands rather than described and hoped for. Its options are literals by
construction — the macro refuses a value that is not one, because it has to write
that value into generated code.
What it does not do is collect. A proc macro sees one item at a time and
nothing gathers the results, so there is no compile-time registry to load; the
register::<T>() chain is that list, written out. The second objection — that
dispatch needs the live function — does not arise at all, since the function was
linked into the binary before the process started.
Discovery makes the registry implicit — it is whatever discovery found. That is the whole point, and it is also the one new failure mode.
What that costs depends on how the job reaches a process that cannot run it.
A worker claims jobs from storage itself, so it can claim one whose task it never registered. That is a fatal, non-retryable failure: retrying cannot make an unregistered task runnable here, so the job dead-letters immediately rather than waiting for a worker that knows it. A worker that picked up eleven of twelve task modules keeps running perfectly for eleven modules' worth of traffic and dead-letters everything belonging to the twelfth.
An attached executor is sent work instead of claiming it, and the scheduler
only ever routes a task name to an executor that advertised it. So the job is
never handed to a peer that cannot run it — it waits for one that can, and if
none attaches before the placement timeout it fails retryably with task '...' was not dispatched: no attached executor advertises it. The symptom is
therefore latency and retries rather than instant dead-lettering, which is
quieter and easier to mistake for a capacity problem.
Either way the jobs belonging to the missing half stop completing, and nothing in the task's own code is wrong.
The first defence is that discovery never swallows an error. Anything that fails to load fails the whole call, loudly, naming what failed:
autodiscover re-raises the ImportError, including for a subpackage — the
standard-library walk would otherwise skip that subtree in silence.
discover() throws TaskDiscoveryError carrying the offending path and the
original failure as cause. Modules are imported one at a time so the first
failure names its own file instead of racing the others.
discover() throws TaskDiscoveryException naming the provider class rather
than contributing nothing and continuing.
Nothing loads at runtime, so there is no load failure to swallow — the link step
already proved every registered task exists. The half that remains is the one
above: a task left off the chain, or one whose chain entry is behind the same
cfg as the task itself.
The fingerprint below will not catch it either. This shell supplies its own handler pool, and a worker that brings a dispatcher reports no registry fingerprint — the column exists to describe what the built-in pool will run, and answering for a pool it does not own would be the wrong answer. So a Rust worker takes no part in the comparison, in either direction.
That covers something that failed to load. It does not cover something that was never reached at all — a stale image, a partial deploy, a directory that did not make it into the container. For that, the scheduler compares peers.
Every executor tells the scheduler which task names it can run when it attaches, because that list is how dispatch routes. The scheduler condenses it into a short value it can compare — the registry fingerprint:
eb5ed0d43c8f2aaa.Two properties are worth knowing. Order and duplicates cannot change it, so two executors that registered the same tasks in different orders agree. And it is derived from the names already on the wire rather than sent alongside them — there is nothing extra to compute, nothing that can disagree with the list it summarises, and an executor written against an older SDK takes part with no changes.
It is not a cryptographic hash. A collision costs a missed warning, never a wrong dispatch.
When an executor attaches, the scheduler compares its fingerprint against the executors already attached. If no live peer advertises the same one, it logs a single warning:
[flexiq] executor python-executor-0198f3d2-4c71-7a3e-b8d5-1f2a9c4e7b60 (python 1.0.0) advertises task registry
eb5ed0d43c8f2aaa, but executor python-executor-0198f3d1-9e05-7c42-a1b7-6d3f08ce5142 (python 1.0.0) advertises
0d49390f67ebbff1; a job for a task only one of them knows fails wherever it lands.
only on python-executor-0198f3d2-4c71-7a3e-b8d5-1f2a9c4e7b60: (none). only on python-executor-0198f3d1-9e05-7c42-a1b7-6d3f08ce5142:
reports.build, reports.export[flexiq] executor node-executor-0198f3d2-4c71-7a3e-b8d5-1f2a9c4e7b60 (node 1.0.0) advertises task registry
eb5ed0d43c8f2aaa, but executor node-executor-0198f3d1-9e05-7c42-a1b7-6d3f08ce5142 (node 1.0.0) advertises
0d49390f67ebbff1; a job for a task only one of them knows fails wherever it lands.
only on node-executor-0198f3d2-4c71-7a3e-b8d5-1f2a9c4e7b60: (none). only on node-executor-0198f3d1-9e05-7c42-a1b7-6d3f08ce5142:
reports.build, reports.export[flexiq] executor java-executor-0198f3d2-4c71-7a3e-b8d5-1f2a9c4e7b60 (java 1.0.0) advertises task registry
eb5ed0d43c8f2aaa, but executor java-executor-0198f3d1-9e05-7c42-a1b7-6d3f08ce5142 (java 1.0.0) advertises
0d49390f67ebbff1; a job for a task only one of them knows fails wherever it lands.
only on java-executor-0198f3d2-4c71-7a3e-b8d5-1f2a9c4e7b60: (none). only on java-executor-0198f3d1-9e05-7c42-a1b7-6d3f08ce5142:
reports.build, reports.exportThis warning is about attached executors, and the Rust SDK does not start
one. WorkerBuilder::spawn starts an in-process worker that claims jobs from
storage rather than attaching, so no process built through this SDK appears on
either side of the comparison. flexiq::ExecutorClient is core's executor door;
the shell does not wrap it.
The failure the warning exists to catch still applies. The registry here is the
register::<T>() chain, so a task that was renamed or removed breaks the build
rather than dead-lettering at runtime — but one that was never added to the
chain compiles fine, and the first job for it fails fatally with task not registered on this worker.
The sdk and version in parentheses are what each executor reported at
attach, so a mixed fleet names them differently on either side of the comparison
— the executor id defaults to <sdk>-executor-<uuid> unless you set one.
The two only on lists are the symmetric difference — exactly the names one
side can run and the other cannot. They are capped at ten per side with a
+N more tail; the fingerprints stay exact, so the cap costs detail and never
the ability to tell two registries apart.
(none) on one side, as above, means that registry is a strict subset of
the other's. That is the shape a half-finished discovery produces, and it is the
case worth acting on fastest.
An executor with no fingerprint — one advertising no tasks at all — takes no part on either side of the comparison: it neither triggers a warning nor satisfies one, because a registry holding nothing is not a registry that differs from everyone else's. That is not the blind spot it sounds like. A discovery that found nothing never reaches the comparison: every SDK that ships an executor refuses to start one on an empty task list ("no tasks are registered on this app, so the executor would never be sent any work"), because it would attach and then sit idle forever. Total failure is loud at startup; it is the partial failure that needs the fingerprint.
The check warns only when no live peer already advertises the joining registry. Rolling fifty workers onto a new task list produces one line — from the first one to attach — not fifty. It needs no memory of what it already warned about, so it behaves the same across a scheduler restart.
A divergent executor still attaches and still receives work. Two registries may differ deliberately — a worker fleet specialised onto one set of tasks is a normal deployment — and refusing the attach would turn a warning into an outage.
only on lists. They name the tasks at risk — the ones only
part of the fleet can run, so they depend on that part staying attached and
having a free slot.task '...' was not dispatched: no attached executor advertises it means the half that knew
them is gone entirely, not merely outnumbered. Those failures are retryable,
so the jobs come back once a peer that advertises the name attaches; check
the dead-letter queue for any that exhausted their retries first, and requeue
them — see
error handling.The comparison covers attached executors, which is where the scheduler sees a registry from more than one peer. In-process workers do not speak the attach protocol, so a single-worker deployment gets no comparison — there is nothing to compare against. The defences that do apply there are the ones above: discovery that never swallows a load failure, and duplicate names that raise instead of overwriting.