Tasks

Binding a handler

A Task<T> is just a name and a payload type — it's a producer-side descriptor, not a function. A worker binds the actual logic separately, as a TaskFunction<T, R> that receives the deserialized payload and returns a result (or null):

try (Worker worker = flexiq.worker()
        .handle(sendEmail, payload -> deliver(payload))             // Task + TaskFunction
        .handle("resize", ImageJob.class, payload -> resize(payload)) // name + payload type
        .start()) { ... }

Handler.of(task, fn) pairs the two for registration via register(handler) or a HandlerRegistry. See Workers for the full worker lifecycle.

Typed tasks from @TaskHandler

Annotate handler methods with @TaskHandler; the compile-time processor (org.byteveda:flexiq-processor) generates a <Class>Tasks companion with a typed Task constant per method plus a bind(...) — name declared once, full generics, zero runtime reflection:

class EmailTasks {
    @TaskHandler("send_email")          // explicit name
    String send(EmailPayload p) { ... }

    @TaskHandler                        // name defaults to "report"
    Report report(List<Metric> metrics) { ... }
}

// generated EmailTasksTasks:
String id = flexiq.enqueue(EmailTasksTasks.SEND, payload);

flexiq.worker()
        .apply(b -> EmailTasksTasks.bind(b, new EmailTasks()))
        .start();

The annotation also accepts queue, maxRetries, timeoutMs, priority, idempotent, and a circuitBreakerThreshold (plus circuitBreakerWindowSeconds/circuitBreakerCooldownSeconds/ circuitBreakerHalfOpenProbes/circuitBreakerHalfOpenSuccessRate) for per-task defaults. See installation for the processor setup.

Each companion also carries a nested Provider that the processor lists in META-INF/services, so a worker can skip the bind(...) line entirely and call .discover() instead — see Task discovery.

Binding a handler

There is no separate descriptor to bind. #[flexiq::task] generates the type and keeps the body, and register::<send_email>() takes that type — so the name, the argument types and the handler cannot drift apart:

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

flexiq::Task is a public trait — NAME, config(), defaults(), run_encoded(), and an optional periodic() — so a task with an unusual shape can implement it by hand and register exactly the same way. The macro is ergonomics, not plumbing.

queue.task("add", (a: number, b: number) => a + b, {
  maxRetries: 3,
  retryBackoff: { baseMs: 1000, maxMs: 60_000 },
  timeoutMs: 30_000,
  maxConcurrent: 4,
  rateLimit: "100/m",
  circuitBreaker: { threshold: 5, windowMs: 60_000, cooldownMs: 30_000 },
});

Beyond per-task config, set defaults for an entire named queue:

queue.configureQueue("emails", { rateLimit: "50/s" });

Fluent methods attach default enqueue options; each returns a new descriptor (the type is immutable):

Task<EmailPayload> sendEmail = Task.of("send_email", EmailPayload.class)
        .queue("emails")
        .priority(5)
        .retries(3)
        .timeout(Duration.ofSeconds(30))
        .delay(Duration.ofSeconds(10));

#[flexiq::task] accepts a closed set of attributes. Anything else is a compile error that lists the accepted ones:

A duration is 500ms, 30s, 5m, 2h, 1d, or a bare unquoted integer of milliseconds — timeout = "30s" and timeout = 30_000 are the same thing. A rate is "<count>/<unit>", the count at least 1 and the unit one of s, m, h (or second, minute, hour): "100/m", "20/s", "1000/h". Both grammars are checked while the macro expands, so a typo is a compile error rather than a worker that starts and quietly never dispatches.

#[flexiq::task(
    name = "emails.send",
    queue = "emails",
    priority = 10,
    max_retries = 5,
    retry_backoff_ms = "2s",
    retry_max_delay_ms = "1m",
    timeout = "60s",
    rate_limit = "100/m",
    max_concurrent = 10,
)]
fn send_email(to: String, subject: String, body: String) -> flexiq::Outcome<()> {
    Ok(())
}

The set splits in two. queue, priority, max_retries, timeout, expires, result_ttl and idempotent are enqueue defaults: they seed every send_email::call(..), and the matching setter on the call overrides them for one job. The rest is scheduler policy, read once per task when the worker starts, with no per-call form.

Use retry_delays for an exact per-attempt wait schedule instead of exponential backoff — the final value repeats for any further retries up to max_retries:

@queue.task(retry_delays=[1, 5, 30])  # 1s after 1st fail, 5s after 2nd, 30s after 3rd
def flaky_api_call():
    ...

retryBackoff only shapes an exponential curve (baseMs, maxMs) — there's no option for an explicit per-attempt delay list.

RetryPolicy.delays(...) is the explicit per-attempt equivalent — applied exactly, with no jitter. Supply at least as many delays as maxRetries; once the list is exhausted, further retries fire immediately:

Task<Order> poll = Task.of("poll", Order.class)
        .retries(3)
        .retryPolicy(RetryPolicy.delays(
                Duration.ofSeconds(5), Duration.ofSeconds(30), Duration.ofMinutes(2)));

The attribute set shapes the exponential curve only — there is no attribute for an exact per-attempt schedule. The field the scheduler reads is RetryPolicy::custom_delays_ms, so a task that needs one implements flexiq::Task by hand and returns it from config():

fn config() -> flexiq::TaskConfig {
    flexiq::TaskConfig {
        retry_policy: flexiq::RetryPolicy {
            max_retries: 3,
            custom_delays_ms: Some(vec![5_000, 30_000, 120_000]),
            ..flexiq::RetryPolicy::default()
        },
        ..flexiq::TaskConfig::default()
    }
}

Explicit delays are applied exactly; the exponential path adds full jitter, so many jobs failing against the same downstream spread out instead of retrying in a wave.

Soft timeouts

A soft timeout raises SoftTimeoutError only when the task cooperatively checks — useful for a long loop that should stop cleanly instead of being killed mid-iteration:

from flexiq import current_job

@queue.task(timeout=300, soft_timeout=60)
def long_running(items):
    for item in items:
        current_job.check_timeout()  # raises SoftTimeoutError if soft_timeout exceeded
        process(item)

See Timeouts for how a timeout is surfaced to middleware and events, and how to make an async task actually stop work when one fires.

See Timeouts — a timeout releases the job's bookkeeping, not the handler's thread, so long-running handlers should poll queue.isCancelRequested(jobId) to actually stop.

There is no soft timeout and no cooperative check: a task body receives its decoded arguments and nothing else, so there is no job handle inside it to poll.

The hard timeout is enforced by the scheduler's stale-job sweep, which fails the job with job timed out after <n>ms and frees its slot — it does not interrupt the handler, which keeps running on its blocking thread until it returns. Bound the work inside the body yourself if that matters: give the HTTP client its own timeout, or break a long loop on a deadline you read at the top.

See Circuit breakers for the full state machine and how to inspect breaker state at runtime.

See Circuit breakers — pair it with retries: retries absorb transient blips, the breaker stops spending that budget on a dependency that's actually down.

Breaker state lives in storage, one row per task, so any process pointed at the same backend can read it. The read is a Storage method rather than a FlexiQ one:

use flexiq::{CircuitState, Storage};

if let Some(row) = queue.storage().get_circuit_breaker("call_external_api")? {
    println!(
        "{:?}, {} failures, opened at {:?}",
        CircuitState::from_i32(row.state),
        row.failure_count,
        row.opened_at
    );
}

list_circuit_breakers() returns every row at once. CircuitState is Closed / Open / HalfOpen, and row.state is the raw i32 those decode from.

max_concurrent caps how many instances of a task run at once, enforced by the scheduler across every worker on shared storage — not per worker:

@queue.task(max_concurrent=3)
def expensive_render():
    ...
# At most 3 instances of expensive_render run simultaneously across all workers.

maxConcurrent limits how many jobs of a task run at once across all workers — enforced by the scheduler against the live running count, so it holds even with many workers on shared storage:

queue.task("transcode", transcode, { maxConcurrent: 2 }); // never more than 2 at once

Apply it to a whole queue instead with queue.configureQueue("video", { maxConcurrent: 4 }). See Concurrency.

Concurrency in the Java SDK is a worker-level setting, not a per-task field on Task: each worker runs its handlers on a thread pool, and the pool's size caps how many jobs that worker executes at once.

Worker worker = queue.worker()
        .handle(transcode, this::transcode)
        .concurrency(2) // at most 2 handlers run at once on this worker
        .start();

To cap a task globally, give it a dedicated queue and size a single worker to the limit, or serialize the critical section with a distributed lock. See Concurrency.

Two caps, at two scopes. max_concurrent is cluster-wide — the scheduler counts live runs of the task across every worker on shared storage, at the cost of a database read. max_in_flight_per_task is this worker's own pool, in-process and free, and stops one slow task from occupying every slot:

#[flexiq::task(max_concurrent = 3, max_in_flight_per_task = 1)]
fn expensive_render(scene: String) -> flexiq::Outcome<()> {
    Ok(())
}

The worker's total is num_workers, set on the builder and defaulting to 4 — that bounds every task together, where the two attributes above bound this one.

Apply middleware to a specific task only, in addition to queue-level middleware:

from flexiq.contrib.sentry import SentryMiddleware

@queue.task(middleware=[SentryMiddleware()])
def important_task():
    ...

Node middleware is registered queue-wide with queue.use(...); there's no per-task middleware option. Scope logic to a task inside the middleware via ctx.taskName in before / after, or toggle a global middleware per task from the dashboard (queue.disableMiddlewareForTask(task, name)).

Java Middleware is registered globally with flexiq.use(...); there's no per-task attachment. Branch inside the middleware on ctx.taskName().

The Rust SDK has no middleware hooks at all, per task or otherwise. The nearest seam is flexiq_core::Worker::on_outcome, a callback handed every processed outcome — success, retry, dead-letter, cancel — which WorkerBuilder does not expose; building the worker through flexiq_core::Worker directly costs you the durable-step fence this shell's pool carries. For behaviour that belongs to one task, put it in the body — the function is your own code and nothing intercepts it on the way in.

Override the queue-level serializer for a specific task — the full round-trip: arguments are serialized with it at enqueue time and deserialized with it on the worker before the task function is called. Both the sync worker and the native async worker honor the per-task serializer, falling back to the queue-level serializer for tasks that have none registered.

from flexiq.serializers import JsonSerializer

@queue.task(serializer=JsonSerializer())
def api_event(payload: dict) -> dict:
    ...

Useful when a task needs a different format (e.g., human-readable JSON for audit tasks) or when the payload isn't picklable. See Serializers.

The serializer is a queue-level choice (new Queue({ serializer })); it can't be swapped per task. For per-task payload handling, register named codecs (compression / encryption / signing) applied on top of the queue serializer:

const queue = new Queue({ dbPath: "myapp.db", codecs: { gzip: new GzipCodec() } });
queue.task("audit_event", handleAudit, { codecs: ["gzip"] }); // transform, not a format swap

See Serializers.

The serializer is a builder-level choice (FlexiQ.builder().serializer(...)); it can't be swapped per task. For per-task payload handling, use Task.codecs(...) — the same layered-transform model (compression / encryption / signing) applied on top of the builder serializer:

Task<AuditEvent> audit = Task.of("audit_event", AuditEvent.class).codecs("gzip");

See Serializers.

The payload codec is the wire CBOR envelope, and it is fixed — there is no serializer to swap, per task or per queue, and no codec layer on top of it. What varies is the types: any argument that is serde::Serialize encodes, and any parameter that is serde::de::DeserializeOwned decodes, so the choice a serializer would give you is made in the function signature instead.

#[derive(serde::Serialize, serde::Deserialize)]
struct AuditEvent {
    tenant: String,
    amount_cents: i64,
}

#[flexiq::task]
fn api_event(event: AuditEvent) -> flexiq::Outcome<()> {
    Ok(())
}

The writer runs in binary mode — is_human_readable() reports false — which is why a type that branches on it, uuid::Uuid being the usual one, round-trips as bytes rather than as a string. Arguments are positional only, since Rust has no keyword arguments, and struct fields keep declaration order: sorting them would change the idempotent dedup key that every language shares.

expires skips a job that wasn't started within the deadline. Set a default on the task, override it per call, or pass it only at enqueue time:

@queue.task(expires=300)  # every job skipped if not started within 5 minutes
def time_sensitive():
    ...

time_sensitive.delay()                      # uses the 300s default
time_sensitive.apply_async(args=(), expires=60)  # tighter window for this call

The Node SDK has no job-expiration deadline. Job options are delayMs (earliest start) and timeoutMs (max run time), but a queued job is never discarded for sitting too long. Gate stale work inside the handler, or cancel it with queue.cancelJob(id).

The Java SDK has no job-expiration deadline. EnqueueOptions offers delay (earliest start) and timeout (max run time), but a queued job is never discarded for sitting too long. Gate stale work inside the handler, or cancel it with flexiq.cancelJob(id).

expires skips a job that has not started within the deadline. Declare a default on the task, or set it for one call with .expires_in_ms(..):

#[flexiq::task(expires = "5m")] // skipped if it has not started within 5 minutes
fn time_sensitive(order_id: String) -> flexiq::Outcome<()> {
    Ok(())
}

queue.enqueue(time_sensitive::call("ord-1".into()))?;                       // the 5m default
queue.enqueue(time_sensitive::call("ord-2".into()).expires_in_ms(60_000))?; // tighter, this call only

Both are relative and resolved against one now, read once at enqueue, so expires_at and a delay_ms on the same call cannot disagree by the width of the call. The sweep that discards an expired job is the scheduler's, so it runs whether or not a worker ever looked at the queue.

By default, tasks are named using module.qualname:

# In myapp/tasks.py
@queue.task()
def process():  # Named: myapp.tasks.process
    ...

Override with an explicit name:

@queue.task(name="my-custom-name")
def process():  # Named: my-custom-name
    ...

Node task names are always explicit — the required first argument of queue.task(name, handler). Unlike Python's module.qualname default, there is no auto-derived name (JS function names are unreliable after bundling).

queue.task("emails.send", (to: string, subject: string) => send(to, subject));
queue.enqueue("emails.send", ["user@example.com", "Welcome"]);

Task.of(name, type) takes an explicit name. With the @TaskHandler annotation, an empty value() defaults the task name to the method name:

@TaskHandler("emails.send")   // explicit name
String send(EmailPayload p) { ... }

@TaskHandler                  // name defaults to the method name: "report"
Report report(List<Metric> m) { ... }

A task is named after its function, and name = "..." overrides that:

// In src/tasks/invoices.rs
#[flexiq::task]
fn process() -> flexiq::Outcome<()> { Ok(()) }        // named: process

#[flexiq::task(name = "my-custom-name")]
fn process_v2() -> flexiq::Outcome<()> { Ok(()) }     // named: my-custom-name

Deliberately not the module path. A path embeds the crate name, so moving the function between modules — or renaming the binary — would change a name a producer in another language has to type verbatim.

Calling a task directly runs it synchronously, bypassing the queue — useful in tests:

result = send_email("user@example.com", "Hello", "World")  # Runs immediately

queue.task() returns the queue, not a task object, so there's nothing to "call directly." Keep a reference to your handler function and invoke it directly in tests; only enqueue() routes through the queue.

const sendEmail = async (to: string, subject: string) => { /* ... */ return "sent"; };
queue.task("sendEmail", sendEmail);

const result = await sendEmail("user@example.com", "Hi"); // runs now, bypasses the queue

A Task<T> carries no logic to call. Invoke the TaskFunction / handler method directly for a synchronous run; flexiq.enqueue(task, payload) is the only path that goes through the queue.

TaskFunction<EmailPayload, String> send = payload -> smtp.deliver(payload);

String result = send.apply(payload); // runs now, bypasses the queue

The macro keeps the original body as run(..), with the function's own signature, so a task is still callable without a queue — which is how you unit test one:

let sent = send_email::run("user@example.com".into(), "Hello".into(), "World".into());

Its error type is Abort, not QueueError: "this attempt failed" and "the queue could not be reached" are different things, so ? deliberately does not bridge them. call(..) mirrors the same signature, which is why a wrong argument to an enqueue is the same compile error as a wrong argument to the function.

Each item can carry its own overrides. See Batch enqueue for enqueue_many(), per-item overrides, and per-item result tracking.

Each entry has its own args and options. See Enqueue options for the full batch-enqueue contract, and Batching for the buffering Batcher helper built on top of it.

Unlike the per-item form above, one shared EnqueueOptions applies to the whole batch — not a list of per-job options. enqueueAll is an alias of enqueueMany. See Batching for the buffering Batcher<T> helper built on top of it.

Every item carries its own options, because each item is a call with the setters on it. What a batch cannot vary is the task: the call's type is the task, so one enqueue_batch is one task name.

A debounce window is refused rather than split out — storage has no batched debounce, and looping it item by item would cost the atomicity that is the reason to send a batch. Enqueue that one on its own. One item whose arguments cannot encode fails the whole call and writes nothing.