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.

Despite the _ms suffix, both backoff attributes take the same duration spellings as timeout — "500ms", "2s", "5m", "2h", "1d" — or a bare integer meaning milliseconds.

max_retries is the one attribute that lands in two places: the task's retry policy and the enqueue defaults. The budget the scheduler actually counts against is the one written on the job row, which is why a per-call .max_retries(..) overrides it cleanly.

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

RetryPolicy::custom_delays_ms replaces the exponential curve with explicit per-attempt delays, honoured exactly and with no jitter. It is not a #[flexiq::task] attribute — the macro rejects it as unknown — so it is reached through a hand-written Task impl, the same shape a circuit breaker needs:

use flexiq::{Task, TaskConfig};

fn config() -> TaskConfig {
    let mut config = <flaky_api_call as Task>::config();
    config.retry_policy.custom_delays_ms = Some(vec![1_000, 5_000, 30_000]);
    config
}

Retry 0 waits 1s, retry 1 waits 5s, retry 2 waits 30s. Once the list runs out, the Full Jitter formula takes over again with base_delay_ms and max_delay_ms as B and M — there is no fire-immediately tail.

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.

There is no filter to configure — no predicate, no type list. The body says what kind of failure it had at the point it fails, by picking a constructor:

#[flexiq::task(max_retries = 5)]
fn charge(order_id: String) -> flexiq::Outcome<String> {
    // Transient: spends the budget and comes back on the backoff curve.
    let response = gateway::charge(&order_id)
        .map_err(|e| flexiq::TaskError::retryable(e.to_string()))?;

    if response.status == 402 {
        // Permanent: dead-letters now, whatever budget is left.
        return Err(flexiq::TaskError::fatal("card declined").into());
    }
    Ok(response.body)
}

Outcome<T> is Result<T, Abort>, and From<TaskError> is the only conversion into Abort — deliberately, because a blanket From<E: Error> would have to guess retryability, the one bit of a failure only the author knows. That is why every fallible call ends in a map_err that decides.

Watch the point-free form: .map_err(TaskError::retryable) compiles only when the error type is already Into<String>, since that is what the constructor takes. For anything else, write the closure and call .to_string() yourself.

A panic is not a third case. catch_unwind catches it and reports a retryable failure carrying the panic's message — a panic is usually transient, a poisoned lock or an unwrap on a row that was briefly absent, and making it the one failure that ignored the retry policy would surprise.

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 "Exception passes filter?" step reads one bool: whether the TaskError the body returned was built by retryable or by fatal. Nothing registered on the task can flip it afterwards, and a panic reaches this step as retryable.

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.

See Timeouts for the timeout attribute and the per-call .timeout_ms(..).

A timeout is detected outside the handler, which is why the pool reports every failure of its own with timed_out: false — nothing inside a running body can tell that its own attempt overran. The scheduler's stale-job reap sets the flag, and it reports the failure as retryable whatever the body would have said, so a timeout always reaches the budget check.

The flag rides the ResultOutcome that flexiq_core::Worker::on_outcome observes; WorkerBuilder does not expose that callback, so reading the flag means building the worker through flexiq_core::Worker directly.

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

There are no middleware hooks and no event bus to subscribe to. The nearest seam is flexiq_core::Worker::on_outcome, a callback over every processed outcome — and WorkerBuilder does not expose it, so reaching it means building the worker through flexiq_core::Worker directly.

Short of that, the history is a storage read rather than a callback:

use flexiq::Storage;

for error in queue.storage().get_job_errors(&job.id, None)? {
    println!("attempt {}: {}", error.attempt, error.error);
}

JobError carries attempt — the retry_count the job held when that attempt failed, so 0 for the first — the error string, which is the canonical {"errtype","message","traceback"} JSON a failed task writes, and a failed_at in Unix milliseconds.

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

The dead-letter reads are on Storage, not on FlexiQ, so queue.storage() is the way in:

use flexiq::Storage;

// The 10 most recent dead letters, newest first.
let dead = queue.storage().list_dead(10, 0, None)?;
for entry in &dead {
    println!("{} {} — {:?}", entry.original_job_id, entry.task_name, entry.error);
}

// Re-enqueue one as a fresh job; the entry is deleted and the new id returned.
let new_job_id = queue.storage().retry_dead(&dead[0].id, None)?;

// Delete entries that failed more than 24 hours ago.
let purged = queue.storage().purge_dead(flexiq::now_millis() - 86_400_000)?;

The third argument of list_dead is the namespace; None spans every one of them. purge_dead takes an absolute Unix-millisecond cutoff, not an age. DeadJob::error is an Option<String> holding the failure's canonical JSON, and a replayed job keeps the original's priority, max_retries, timeout_ms and result_ttl_ms — the entry stored all four.

See Dead-letter queue.