Error Handling

There's no in-task "stop retrying" signal from inside a failing attempt — set retries to 0 for a fire-once task.

There's no in-task "stop retrying" signal from inside a failing attempt — set retries to 0 for a fire-once task.

TaskFunction declares throws Exception, so checked exceptions propagate without wrapping. maxRetries defaults to 0 — never retry — so a fresh task is fire-once until you opt into retries.

A failing attempt can also decide for itself: throwing NonRetryableException dead-letters the job immediately, RetryableException insists on a retry. See Typed signals from the handler.

A failing attempt does decide for itself, and it is the only thing that does: TaskError::fatal(..) dead-letters the job at once whatever budget is left, TaskError::retryable(..) spends one. Set max_retries = 0 when the whole task should be fire-once rather than one failure inside it. See Retries.

The scheduler detects an exceeded timeout externally — its maintenance reaper checks roughly every 5 seconds — and marks the job failed, which triggers the retry/DLQ logic above.

The dispatcher races the task against the timeout (tokio::time::timeout). A timeout fails the attempt but cannot stop the handler, which keeps running — and the job's retry can start beside it and repeat its side effects. It does not abort the task's AbortSignal either — that fires only on a cancel request. Give the work its own deadline, shorter than timeoutMs, and combine it with the task's signal — AbortSignal.any([signal, AbortSignal.timeout(ms)]), on Node 20.3 or later — so it stops on a cancel too and cleans up before the attempt is failed. Keep its side effects idempotent. See Timeouts for an example.

The dispatcher races the handler against the timeout. Unset (or 0) means no limit.

The scheduler detects an exceeded timeout externally — its maintenance reap runs every reap_interval polls, roughly every 5 seconds at the default SchedulerConfig — and marks the job failed, which triggers the retry/DLQ logic above. Nothing races the handler.

Unset is not "no limit" here: timeout_ms defaults to 300_000, five minutes, on every enqueue, and .timeout_ms(..) overrides it per call.

For CPU-bound synchronous work that might hang, honoring AbortSignal isn't enough on its own — see Troubleshooting: worker is unresponsive for offloading it.

Make long-running handlers cooperative: poll flexiq.isCancelRequested(jobId) or bound their own blocking calls, so the underlying work actually stops. See Cancellation.

Nothing preempts a blocking thread, so make a long-running body cooperative: bound its own blocking calls, and poll Storage::is_cancel_requested(&job.id, None) between units of work. That needs the Job, which a macro-expanded body never receives — see Cancellation for the hand-written Task impl that does.

Soft timeouts

While the hard timeout above is enforced by the scheduler from the outside, a soft timeout lets the task itself react to time pressure — it's cooperative, so the task must call check_timeout() at safe points:

from flexiq import current_job

@queue.task(soft_timeout=30)
def long_task():
    for chunk in data_chunks:
        process(chunk)
        current_job.check_timeout()  # raises SoftTimeoutError after 30s

There is no soft timeout

Only the hard one, enforced by the scheduler from outside. A body that wants to react to time pressure measures it itself and ends the attempt on its own terms:

use std::time::{Duration, Instant};

#[flexiq::task(timeout = "60s")]
fn long_task(chunks: Vec<String>) -> flexiq::Outcome<()> {
    let deadline = Instant::now() + Duration::from_secs(30);
    for chunk in chunks {
        process(&chunk);
        if Instant::now() > deadline {
            return Err(flexiq::TaskError::retryable("soft deadline passed").into());
        }
    }
    Ok(())
}

That spends a retry, exactly as the hard timeout would. Use TaskError::fatal instead when overrunning means the work will never fit.

A job handle also exposes its own history directly:

job = unreliable_task.delay()

for error in job.errors:
    print(f"Attempt {error['attempt']}: {error['error']} at {error['failed_at']}")

There is no job handle to hang a history off — Job is the row, read back by id. The most recent failure is on the row itself; the whole history is the storage read:

use flexiq::Storage;

if let Some(job) = queue.get_job(&job_id)? {
    println!("last error: {:?}", job.error);  // the canonical TaskError JSON
}

for entry in queue.storage().get_job_errors(&job_id, None)? {
    println!("attempt {}: {} at {}", entry.attempt, entry.error, entry.failed_at);
}

entry.attempt is the retry_count the job carried when that attempt failed, so the first one is 0.

See Retries for the full dead-letter queue API — listing, purging, and error history.

See Retries for the backoff curve and the retryable/fatal choice, and Dead-letter queue for the Storage calls that list, replay and purge entries — none of them is on FlexiQ.

See Cancellation for cancelling a still-pending job and progress reporting.

See Cancellation for cancelling a still-pending job, progress reporting, and getting the job id into a handler.

See Cancellation for cancelling a still-pending job, and for getting the job id into a body that has to poll for the flag — #[flexiq::task] hands a body its decoded arguments and nothing else.

See Middleware for the full hook lifecycle and execution order.

See Middleware and Events for the full hook and event lists.

See Middleware and Events for the full hook and event lists.

See Middleware for what flexiq_core::Worker::on_outcome reaches and what it costs to route a worker through it, and Events for why there is no stream to subscribe to.

Serialization errors

The default SmartSerializer uses MessagePack for common types and falls back to cloudpickle for anything else. The cloudpickle fallback fails on:

• Unpicklable objects: open file handles, database connections, sockets, thread locks
• Module-level objects that can't be resolved: dynamically generated classes without stable import paths
• Very large objects: cloudpickle has no hard limit, but extremely large payloads slow down SQLite writes

Fix: pass simple, serializable data (strings, numbers, dicts, lists) as task arguments, and reconstruct complex objects inside the task:

# Bad — passing a connection object
@queue.task()
def query(conn, sql):  # conn can't be pickled
    return conn.execute(sql)

# Good — pass connection info, create inside the task
@queue.task()
def query(db_url, sql):
    conn = create_connection(db_url)
    return conn.execute(sql)

See Serialization for the full serializer list, including JsonSerializer for cross-language or untrusted payloads.

Exception filtering

Use retry_on and dont_retry_on to control which exceptions trigger retries:

@queue.task(
    max_retries=5,
    retry_on=[ConnectionError, TimeoutError],
    dont_retry_on=[ValueError],
)
def call_api(url):
    resp = requests.get(url, timeout=10)
    resp.raise_for_status()
    return resp.json()

See Retries — Exception Filtering for details.

Test mode for isolation

Run tasks synchronously and inspect errors without a worker using test mode:

with queue.test_mode() as results:
    risky_task.delay()

    if results[0].failed:
        print(results[0].error)
        print(results[0].traceback)

Encoding errors

There is no serializer to choose and no fallback codec: a payload is the wire CBOR envelope, fixed. What can still fail is a value with no representation in that envelope, and each direction has its own type.

An encoding failure is carried on the TaskCall rather than raised by call(..) — that call mirrors the task's own signature and has nowhere to put a Result, and a batch stays one expression. It surfaces from enqueue, naming the task:

let err = queue.enqueue(takes_a_big_number::call(u64::MAX)).unwrap_err();
// task `takes_a_big_number` could not encode its arguments: …

A decoding failure happens on the worker instead, and it is fatal — no amount of retrying makes a payload decodable — so the job dead-letters with a message naming the task.

Fixing it

Pass values the envelope can hold, and reconstruct anything else inside the body — a task takes its arguments and nothing else, so a connection or a handle was never going to travel anyway:

// Bad: a value past i64::MAX has no representation, and `enqueue` says so.
#[flexiq::task]
fn count_rows(upper_bound: u64) -> flexiq::Outcome<u64> {
    Ok(upper_bound)
}
queue.enqueue(count_rows::call(u64::MAX))?;   // Err, naming the task

// Good: pass what identifies the work, and build the rest in the body.
#[flexiq::task]
fn query(db_url: String, sql: String) -> flexiq::Outcome<()> {
    let retry = |e: std::io::Error| flexiq::TaskError::retryable(e.to_string());
    let conn = connect(&db_url).map_err(retry)?;
    conn.execute(&sql).map_err(retry)?;
    Ok(())
}

Testing a task without a worker

#[flexiq::task] leaves the body callable: query::run(..) is the original function with its original signature, so a unit test exercises it with no queue, no storage and no worker. FlexiQ::in_memory() is the other half — a private database that dies with the handle — when the test needs an enqueue as well.

All FlexiQ exceptions inherit from FlexiQError, so you can catch the base class for broad handling:

FlexiQError (base)
├── TaskTimeoutError        — hard timeout exceeded
├── SoftTimeoutError        — soft timeout exceeded (check_timeout)
├── TaskCancelledError      — task cancelled (check_cancelled)
├── MaxRetriesExceededError — all retry attempts exhausted
├── SerializationError      — serialization/deserialization failure
├── CircuitBreakerOpenError — circuit breaker is open
├── RateLimitExceededError  — rate limit exceeded
├── JobNotFoundError        — job ID not found (also a KeyError)
└── QueueError              — queue-level operational error

from flexiq import FlexiQError, SoftTimeoutError, TaskCancelledError

try:
    result = job.result(timeout=30)
except FlexiQError as e:
    print(f"FlexiQ error: {e}")

Import any exception directly from the flexiq package:

from flexiq import TaskCancelledError, SoftTimeoutError, SerializationError

Every SDK error extends FlexiQError, so you can catch the base class for broad handling:

import { FlexiQError, JobFailedError } from "@byteveda/flexiq";

try {
  const result = await queue.result(jobId, { timeoutMs: 30_000 });
} catch (err) {
  if (err instanceof FlexiQError) {
    console.error("FlexiQ error:", err.message);
  }
}

Every SDK error is an unchecked FlexiQException; the errors package narrows it so callers can catch exactly what they care about:

try {
    queue.enqueue(FETCH, url);
} catch (FlexiQException e) {
    log.error("FlexiQ error: {}", e.getMessage());
}

There is no hierarchy and no base class to catch. Every operation returns flexiq::Result<T>, which is Result<T, QueueError> — one enum, so a match on it is exhaustive. The variants a caller usually names:

TaskError is the other error type in play and is deliberately not one of these. A QueueError means the queue could not be reached; a TaskError means the work failed. ? does not bridge them, and a task body's error type is Abort, not QueueError, for exactly that reason.