Retries

Coming from BullMQ? Watch the off-by-one: BullMQ's attempts counts the total tries including the first (attempts: 3 = 1 try + 2 retries). flexiq's maxRetries counts only the retries after the first (maxRetries: 3 = 1 try + 3 retries = 4 total). BullMQ's backoff: { type: "exponential", delay } maps to retryBackoff.baseMs.

retry_delays sets each attempt's base delay by index instead of computing it from the exponential curve:

@queue.task(max_retries=5, retry_delays=[1.0, 5.0, 30.0])
def flaky_api_call(url):
    ...

Retry 0 waits ~1s, retry 1 waits ~5s, retry 2 waits ~30s — each still gets jitter up to retry_backoff seconds added on top (retry_backoff isn't overridden, just not used to compute the listed delays). Once the list runs out (retries 3 and 4 here), flexiq falls back to the exponential formula above using retry_backoff and max_retry_delay as B and M.

RetryPolicy.delays(...) replaces the exponential curve with explicit per-attempt delays, applied exactly with no jitter:

Task<Order> CHARGE = Task.of("charge", Order.class)
        .maxRetries(3)
        .retryPolicy(RetryPolicy.delays(
                Duration.ofSeconds(1),
                Duration.ofSeconds(5),
                Duration.ofSeconds(30)));

Supply at least maxRetries delays — once the list is exhausted, further retries fire immediately (no exponential fallback).

Node has no per-attempt delay list — tune the exponential curve instead via retryBackoff (baseMs is B, maxMs is M in the backoff formula above):

queue.task("flakyApiCall", handleCall, {
  maxRetries: 5,
  retryBackoff: { baseMs: 2000, maxMs: 300_000 }, // 2s base, 5min cap
});

Control which exceptions trigger a retry with retry_on (a whitelist) or dont_retry_on (a blacklist) — set one, not both:

# Whitelist: retry only these exceptions; everything else skips straight to DLQ.
@queue.task(max_retries=5, retry_on=[ConnectionError, TimeoutError])
def fetch_data(url):
    response = requests.get(url)
    response.raise_for_status()
    return response.json()

# Blacklist: retry everything except these exceptions.
@queue.task(max_retries=5, dont_retry_on=[ValueError])
def parse_data(raw):
    return json.loads(raw)  # a ValueError here is permanent — don't retry it

If neither is set, every exception triggers a retry. If both are set, dont_retry_on is checked first — a match there skips retry regardless of retry_on — and retry_on then acts as the whitelist for everything else. In practice, use one or the other.

Coming from Celery: retry_backoff here is a float — the base delay in seconds — not Celery's boolean flag. Celery's autoretry_for=(SomeError,) maps to flexiq's retry_on=[SomeError].

retryOn classifies a thrown error: return false and the job dead-letters immediately, whatever budget is left. One predicate covers both directions — test for the errors worth retrying, or negate a test for the permanent ones:

// Whitelist: retry only these; everything else skips straight to DLQ.
queue.task("fetchData", fetchData, {
  maxRetries: 5,
  retryOn: (error) => error instanceof ConnectionError || error instanceof TimeoutError,
});

// Blacklist: retry everything except a permanent parse failure.
queue.task("parseData", parseData, {
  maxRetries: 5,
  retryOn: (error) => !(error instanceof SyntaxError),
});

The predicate is synchronous — the job cannot settle until it answers — and runs on the worker that caught the error, so it sees the live Error object, not just its type name. One that throws is treated as true: a broken classifier must not silently turn transient failures into dead letters.

It sees every error raised while running the task, not only the handler's: a before/after middleware hook or result serialization can fail too, and a whitelist like the first example above dead-letters those as well. Payload decoding fails earlier and always retries, as does a timeout.

retryOn classifies a thrown exception: return false and the job dead-letters immediately, whatever budget is left. One predicate covers both directions — test for the exceptions worth retrying, or negate a test for the permanent ones:

// Whitelist: retry only these; everything else skips straight to DLQ.
Task<String> FETCH = Task.of("fetch", String.class)
        .maxRetries(5)
        .retryOn(error -> error instanceof IOException);

// Blacklist: retry everything except a permanent parse failure.
Task<String> PARSE = Task.of("parse", String.class)
        .maxRetries(5)
        .retryOn(error -> !(error instanceof IllegalArgumentException));

Unlike the backoff curve, the predicate never reaches the scheduler — it runs on the worker that caught the exception, so it sees the live Throwable. One that throws is treated as true: a broken classifier must not silently turn transient failures into dead letters.

It sees every exception raised while running the task, not only the handler's: before/after middleware, payload decoding and result serialization can fail too, and a whitelist like the first example above dead-letters those as well. A timeout is detected outside the handler and always retries.

Typed signals from the handler

A predicate classifies a task's failures by type up front. When only the handler knows — the same IOException is transient on a 503 and permanent on a 422 — throw the intent instead:

queue.worker().handle(CHARGE, (Order order) -> {
    Response response = gateway.charge(order);
    if (response.status() == 402) {
        throw new NonRetryableException("card declined");  // dead-letters now
    }
    if (response.status() >= 500) {
        throw new RetryableException("gateway " + response.status());  // spends the budget
    }
    return response.body();
});

Both live in org.byteveda.flexiq.errors and both beat retryOn — the throw site is more specific than the task-wide predicate, so it wins. They are honoured through the cause chain too, so a signal wrapped by framework code still counts; if a chain carries both, the outermost wins.

The "Exception passes filter?" step only applies filtering when retry_on or dont_retry_on is set on the task — otherwise every exception passes straight through to the retry-budget check.

The "Exception passes filter?" step only applies filtering when retryOn is set on the task — otherwise every error passes straight through to the retry-budget check.

The "Exception passes filter?" step is answered by RetryableException / NonRetryableException when the handler threw one; otherwise it applies retryOn, and with no predicate every exception passes straight through to the retry-budget check.

The scheduler reaps timed-out jobs on a periodic sweep (checking every ~5 seconds):

@queue.task(timeout=10)  # 10 second timeout
def slow_task():
    time.sleep(60)  # Will be reaped after 10s

The timed_out flag drives the on_timeout middleware hook — it fires in addition to on_retry / on_dead_letter whenever the failure was a timeout, so you can alert on wedged tasks separately from ordinary errors.

See Timeouts for timeoutMs, how the dispatcher enforces it, and the timedOut: true flag surfaced to the onRetry / onDeadLetter middleware hooks and job.retrying / job.dead events.

See Timeouts for .timeout(...), how the dispatcher enforces it, and the timed-out flag surfaced as OutcomeEvent.timedOut to event listeners and middleware outcome hooks.

Via middleware, or the job.retrying / job.dead events:

from flexiq import EventType, TaskMiddleware

class AlertOnRetry(TaskMiddleware):
    def on_retry(self, ctx, error, retry_count):
        log.warning(f"{ctx.task_name} retrying (attempt {retry_count}): {error}")

    def on_dead_letter(self, ctx, error):
        alert_ops(ctx.task_name, error)

queue.add_middleware(AlertOnRetry())

# Or, equivalently, as events:
queue.on(EventType.JOB_RETRYING, lambda e: log.warning(f"retrying {e}"))
queue.on(EventType.JOB_DEAD, lambda e: alert_ops(e))

The job.retrying event fires on each retry; once the budget is exhausted the job moves to the dead-letter queue and job.dead fires:

queue.on("job.retrying", (e) => log.warn(`retrying ${e.taskName}`));
queue.on("job.dead", (e) => alertOps(e));

Observe retries as they happen via middleware onRetry or a worker event listener; once the budget is exhausted the job moves to the dead-letter queue and the DEAD event fires:

Worker worker = queue.worker()
        .handle(CHARGE, this::charge)
        .on(EventName.RETRY, event -> log.info("retrying {}", event.taskName))
        .on(EventName.DEAD, event -> alertOps(event))
        .start();

# List the 10 most recent dead letters
dead = queue.dead_letters(limit=10, offset=0)

for d in dead:
    print(f"Job: {d['original_job_id']}")
    print(f"Task: {d['task_name']}")
    print(f"Error: {d['error']}")
    print(f"Retries: {d['retry_count']}")

# Re-enqueue a dead letter job (creates a new job)
new_job_id = queue.retry_dead(dead[0]["id"])

# Delete dead letters older than 24 hours
deleted = queue.purge_dead(older_than=86400)

Every failed attempt is recorded with its error message, accessible via job.errors:

job = unreliable.delay()

# After the job exhausts all retries...
for error in job.errors:
    print(f"Attempt {error['attempt']}: {error['error']}")

max_retries counts retries after the first attempt — max_retries=3 means the task runs up to 4 times total, recording 4 entries in job.errors (attempts 0–3). Each entry has id, job_id, attempt (0-indexed), error, and failed_at (Unix ms).

See Dead-letter queue for deadLetters(), retryDead(), purgeDead(), and per-attempt error history via getJobErrors().

See Dead-letter queue for listDead(), retryDead(), purgeDead(), and per-attempt error history via jobErrors().