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.

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));

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)));

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.

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.

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.

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().

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.

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).

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) { ... }

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

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.