Middleware

The Rust SDK is not one of them: there are no hooks, and nothing to register. The nearest seam is flexiq_core::Worker::on_outcome, which the rest of this page keeps returning to.

Subclass TaskMiddleware and override only the hooks you need — unimplemented hooks default to a no-op on the base class.

Middleware is a plain object whose hooks are all optional — implement only what you need.

Middleware is an interface whose hooks are all no-op default methods — override only what you need.

WorkerBuilder has no slot for any of this. Its whole surface is register::<T>(), queues, num_workers, worker_id, scheduler_config and spawn — tasks and scheduler settings, with nowhere a hook could go.

TaskMiddleware exposes 8 hooks:

ctx is a JobContext — the same object as current_job — exposing ctx.id, ctx.task_name, ctx.retry_count, and ctx.queue_name.

on_timeout fires when the maintenance reaper detects a stale job that exceeded its hard timeout — before on_retry (if it will be retried) or on_dead_letter (if retries are exhausted), so you can react to the timeout itself independently of the job's eventual fate.

Middleware exposes 9 hooks:

The execution hooks (before/after/onError) receive a TaskContext with taskName, jobId, and args. The outcome hooks receive an OutcomeEvent with taskName, jobId, queue, retryCount, timedOut (which separates a timeout from other failures), and durationMs (absent when nothing measured the run).

Middleware exposes 9 hooks:

The execution hooks receive a TaskContext with taskName, jobId, a mutable per-execution attributes() map shared across a job's hooks, job() with lazily-loaded metadata, and elapsedMs() — time spent on this execution so far, so after/onError needn't start their own timer. The outcome hooks receive an OutcomeEvent with taskName, jobId, error, retryCount, timedOut (which separates a timeout from other failures), and durationMs() (null when nothing measured the run).

Zero hooks. What on_outcome gives you instead is one callback over the ResultOutcome enum, whose arms line up with the back half of a hook list and nothing at all with the front:

A seventh arm has no counterpart at all: Superseded { job_id } is a result the job has already moved past — a reclaimed owner finishing late — and reporting it would double-count. ResultOutcome is #[non_exhaustive], so a match on it outside the core crate needs a trailing _ arm — the compiler will not let you enumerate the variants, which is how a later outcome gets ignored instead of breaking your build.

Keys present in options: priority, delay, queue, max_retries, timeout, unique_key, metadata.

EnqueueContext also exposes mutable args, and throwing from onEnqueue aborts the enqueue. This is the same producer-side seam as enqueue interceptors, which run first and can convert, redirect, or reject the call outright before onEnqueue sees it.

EnqueueContext also exposes payload()/payload(...) to replace the job's payload outright, and a mutable metadata() map that travels with the job (readable at execution via context.job().metadata()). When the options already carry metadata that is a JSON object, the hook's map is layered over it (hook keys win) instead of replacing it. Throwing from onEnqueue aborts the enqueue. For typed convert/redirect/reject decisions, use an interceptor instead — interceptors run before middleware onEnqueue.

The other place a default can live is the declaration. Task::defaults() returns an EnqueueOptions seeded from the #[flexiq::task] attribute, and every TaskCall starts from it — so queue, priority, max_retries and timeout set on the attribute already apply to every call without a wrapper. What there is no API for is the reverse: a prebuilt EnqueueOptions cannot be handed to a TaskCall that already exists. Overrides are per-call setters, or a helper function like the one above.

1. Global middleware (registered via Queue(middleware=[...])) runs first
2. Per-task middleware (via @queue.task(middleware=[...])) runs second
3. Within each group, middleware runs in registration order
4. after() and the Rust-dispatched outcome hooks run in that same forward order, not reversed — after() only fires for middleware whose before() didn't raise.

Every hook — before, after, onError, and the outcome hooks — runs in the same registration order for every middleware; the Node SDK doesn't reverse after the way an "onion" middleware model would.

Multiple middlewares run in registration order for every phase — before, after, onError, and the outcome hooks are not reversed.

Nothing to order: there is one callback, and calling on_outcome twice replaces it rather than appending. The only sequence that exists is the worker's — the body runs on a blocking thread, its JobResult goes down a channel, and the drain thread hands the scheduler a batch of them and calls the callback once per outcome the scheduler returned, in that order.

If a middleware hook raises an exception:

• before(): the exception is logged, but subsequent middleware before() hooks still run. The task executes normally.
• after() / on_sleep(): the exception is logged. The other hooks in the chain still run.
• on_retry() / on_dead_letter() / on_timeout() / on_cancel(): logged and swallowed — these are notification hooks, not control flow.

Middleware exceptions never prevent task execution or result handling.

The execution hooks (before/after/onError) are awaited in the hot path: a rejected before or after fails the attempt like a thrown handler error. onError and onSleep are each wrapped per middleware, so a throwing onError never masks the original task failure and a throwing onSleep never un-commits a sleep that already happened; both are logged at debug. The outcome hooks (onCompleted/onRetry/onDeadLetter/ onCancel) are isolated per middleware — one that throws or rejects is logged at debug and doesn't stop the rest from running.

The execution hooks run inside the attempt — a before or after that throws fails the attempt like a handler error, and is caught by the same onError path. onError and onSleep are isolated per middleware instead: one that throws is caught and logged, because escaping there would leave the dispatch with no outcome to report at all. Outcome hooks are isolated for the same reason.

The two sides are not isolated the same way, and the difference matters.

A panic inside a task body is contained: the shell's pool catches it and records a retryable failure naming the task and the panic message, and the blocking thread is reused afterwards. A panic inside an on_outcome callback is not caught by anything — it unwinds the worker's drain thread, and results stop being handled from that point on. Treat the callback as code that must not panic: no unwrap on anything the outcome did not already hand you.

The hook keeps its thread: a middleware may push a thread-local scope in before for the handler to run inside, so moving the call somewhere it could be abandoned would break it. The budget therefore reports an overrun rather than ending it — the log names the middleware and the hook, and the attempt stays blocked until the hook returns. Under the prefork pool the attempt is bounded regardless: a child that outruns its task's timeout is killed, hooks included.

Past the budget the chain stops awaiting the hook and moves on. Nothing can cancel a promise, so the hook itself keeps running; what ends is the attempt's wait for it. A rejection that arrives after that point is logged at debug.

The hook keeps the dispatch thread — a middleware may open a thread-bound scope in before for the handler and close it from after — so the budget is enforced by interrupting that thread. Blocking that honours interrupts (Future.get, awaitTermination, a socket with a timeout) ends there; anything that ignores them runs on, and the overrun is logged either way. The interrupt flag is cleared before the handler runs, so it never inherits one.

There is no budget to set, because there is nothing in the attempt to bound but the body itself — and a task's timeout does not interrupt that either. Nothing cancels a running handler: the deadline is noticed by the stale-job reap, which records the timeout and lets the job retry while the abandoned thread keeps its slot until the body returns.

The unbounded thing to watch is the callback. on_outcome is one closure shared by every job on the worker, run on the single drain thread, with no deadline around it — a call that blocks there does not slow one attempt down, it stops the worker recording any outcome at all. Hand slow work to a channel and return.

Apply middleware to a specific task using the middleware parameter on @queue.task:

@queue.task(middleware=[MetricsMiddleware()])
def process(data):
    ...

Per-task middleware runs after global middleware, in registration order.

There's no separate per-task registration — every middleware passed to queue.use() applies to every task. Scope a middleware's own hooks to specific tasks by checking ctx.taskName inside them, or by giving the middleware its own task-name filter (as the OpenTelemetry integration does with taskFilter).

There's no separate per-task registration — every middleware passed to flexiq.use() applies to every task. Scope a middleware's own hooks to specific tasks by checking context.taskName inside them.

No per-task registration either, and nowhere to put one: #[flexiq::task] accepts a fixed key set and an unrecognised key is a compile error that lists the ones it takes. Scoping happens inside the single callback, by matching on task_name before doing anything — every arm except Superseded carries it.

Middleware vs hooks

Coming from Celery signals? Use these instead of connecting to task_prerun / task_postrun / task_failure: queue-level hooks for global behavior, or per-task middleware for scoped behavior.

flexiq has two systems for running code around tasks:

Middleware runs inside the task wrapper (closer to the task function), while hooks run outside. In practice, middleware before() fires first, then before_task hooks. On completion, on_success/on_failure hooks fire, then middleware after().

See the OpenTelemetry guide for setup details.

See the OpenTelemetry and Sentry integration guides for setup details.

See the Micrometer and Sentry integration guides for setup details.

There are no integration guides to point at, because there are no integrations. Monitoring covers the two real sources — the callback and queue.stats() — Events documents every field on ResultOutcome and what taking that route costs a task with durable steps, and Logging covers what the engine already emits without any of this.