Sentry Integration

Report task failures to Sentry, tagged by task name, job id, and queue.

Where a binding ships a Sentry middleware, it reports task failures to Sentry, tagging each event with the task name, job id, and queue so failures are groupable and searchable. Sentry itself is never initialized for you — do that in your own app first.

Installation

Install the sentry extra, which pulls in sentry-sdk:

bash

pip install flexiq[sentry]

Peer dependency: install @sentry/node yourself and import the middleware from the flexiq/contrib/sentry subpath.

bash

npm install @sentry/node

Optional dependency: the SDK compiles against io.sentry:sentry as compileOnly, so nothing lands on your classpath until you add the runtime dependency yourself.

kotlin

implementation("io.sentry:sentry:7.14.0")

Nothing to install from this side: there is no Sentry Cargo feature and no optional dependency behind one. Add the sentry crate to your own Cargo.toml and initialize it in main, exactly as you would in a binary with no queue in it at all.

Initialize Sentry, then wire it in:

python

import sentry_sdk
from flexiq import Queue
from flexiq.contrib.sentry import SentryMiddleware

sentry_sdk.init(dsn="https://examplePublicKey@o0.ingest.sentry.io/0")

queue = Queue(db_path="myapp.db", middleware=[SentryMiddleware()])

import * as Sentry from "@sentry/node";
import { sentryMiddleware } from "@byteveda/flexiq/contrib/sentry";

Sentry.init({ dsn: process.env.SENTRY_DSN });

queue.use(sentryMiddleware());

java

import io.sentry.Sentry;
import org.byteveda.flexiq.contrib.SentryMiddleware;

Sentry.init(options -> options.setDsn(System.getenv("SENTRY_DSN")));

flexiq.use(new SentryMiddleware());

The Rust SDK ships no Sentry integration, and has no middleware to hang one on. Initialize the sentry crate in your own main and report from inside the task body, which is the only place the real error still exists: by the time a failure reaches the queue it is a TaskError — a message and a retryable flag, recorded as {"errtype": "TaskError", "message": …, "traceback": []} — with the error's type and any backtrace already gone.

rust

#[flexiq::task(max_retries = 3)]
fn process_payment(order_id: String, cents: i64) -> flexiq::Outcome<i64> {
    charge(&order_id, cents).map_err(|err| {
        // report `err` here, while it is still a typed error
        flexiq::TaskError::retryable(err.to_string())
    })?;
    Ok(cents)
}

No binding initializes Sentry for you — configure the DSN, environment, and sampling with the Sentry SDK itself. Where a middleware ships, its hooks are safe no-ops until Sentry has been initialized.

What it captures

The bindings differ on when a failure is reported:

Every task execution opens a Sentry scope tagged with the job's metadata (prefix customizable via tag_prefix):

Tag	Value
`flexiq.task_name`	The registered task name
`flexiq.job_id`	The job ID
`flexiq.queue`	The queue name
`flexiq.retry_count`	Current retry attempt

The Sentry transaction name defaults to flexiq:<task_name> (customizable via transaction_name_fn). When a task raises, SentryMiddleware calls sentry_sdk.capture_exception() automatically — every failed attempt is reported, not just the terminal failure. When a task is retried, a breadcrumb is added (category flexiq by default, level warning, message Retrying <task_name> (attempt <N>): <error>), so the final failure carries a trail of the retries leading up to it.

The exception (with its stack trace) is captured internally in onError and held per job. It's only reported to Sentry once the job dead-letters — so each dead job produces a single event carrying the original stack plus the task/job/queue tags. Successful jobs and jobs that recover on retry are never reported. Set captureRetries: true to also report each intermediate retry as a "warning"-level event.

Tags set on every event: flexiq.task_name, flexiq.job_id, flexiq.queue, flexiq.retry_count, and flexiq.timed_out (when the job timed out). If a job dead-letters from a timeout without throwing, a synthetic error is captured so the failure is still recorded.

Two kinds of events are reported:

Every failed attempt — the handler's exception is captured with its stack trace as soon as it throws. A job that recovers on a later retry still produces one event per failed attempt.
Dead-letters — when a job exhausts its retries, a FATAL-level message (task dead-lettered: <error>) marks the terminal failure.

Both event kinds carry the tags flexiq.task (task name) and flexiq.job (job id).

Nothing is captured for you: what reaches Sentry is whatever the task body reports, on whichever attempts run it. Nor are the task name, job id and queue tagged for you — a task body receives its arguments and nothing else, so anything you want on the event has to come from your own capture call.

Configuration

python

SentryMiddleware(
    tag_prefix="myapp",
    transaction_name_fn=lambda ctx: f"task-{ctx.task_name}",
    task_filter=lambda name: not name.startswith("internal."),
    extra_tags_fn=lambda ctx: {"worker.host": socket.gethostname()},
)

sentryMiddleware({
  tagPrefix: "myapp",
  captureRetries: true,
  level: "error",
  extraTags: (event) => ({ "worker.host": os.hostname() }),
  taskFilter: (taskName) => !taskName.startsWith("internal."),
});

java

// The one-argument constructor takes a Predicate<String> over the task
// name; return false to skip a task entirely.
new SentryMiddleware(task -> !task.startsWith("noisy."));

Nothing to configure: with no middleware there is no tag prefix, no capture level, and no task filter. The filter is wherever you put the reporting call — a body that does not report is a task that is not reported — and a callback over flexiq::ResultOutcome can match on task_name before it says anything.

Parameter	Type	Default	Description
`tag_prefix`	`str`	`"flexiq"`	Prefix for Sentry tag keys and breadcrumb category.
`transaction_name_fn`	`Callable[[JobContext], str] \| None`	`None`	Custom transaction name builder. Receives `JobContext`. Defaults to `<prefix>:<task_name>`.
`task_filter`	`Callable[[str], bool] \| None`	`None`	Predicate on task name. Return `True` to report, `False` to skip. `None` reports all tasks.
`extra_tags_fn`	`Callable[[JobContext], dict[str, str]] \| None`	`None`	Returns extra Sentry tags to set. Receives `JobContext`.

Option	Type	Default	Description
`tagPrefix`	`string`	`"flexiq"`	Prefix for Sentry tag keys.
`captureRetries`	`boolean`	`false`	Also report each retried failure as a `"warning"` event.
`level`	`SeverityLevel`	`"error"`	Severity for the terminal dead-letter event.
`extraTags`	`(event) => Record<string, string>`	—	Extra tags merged onto the event.
`taskFilter`	`(taskName) => boolean`	—	Return `false` to skip a task.

SentryMiddleware has no other options beyond the task filter — DSN, environment, and sampling are all configured through the Sentry SDK itself.

The two options those tables keep offering have plain equivalents here rather than parameters. The task filter is the placement of the reporting call: a body that never reports is a task that is never reported. The tags are whatever you attach at the capture site — and what is in scope there is the task's own arguments, nothing more. There is no ambient job context in a Rust task body, so a job id on the event has to come from an argument you put there yourself, or from an on_outcome callback where the outcome carries it.

Full example

python

import sentry_sdk
from flexiq import Queue
from flexiq.contrib.sentry import SentryMiddleware

# Initialize Sentry first
sentry_sdk.init(
    dsn="https://examplePublicKey@o0.ingest.sentry.io/0",
    traces_sample_rate=1.0,
)

# Create queue with Sentry middleware
queue = Queue(db_path="myapp.db", middleware=[SentryMiddleware()])

@queue.task(max_retries=3)
def process_payment(order_id: str, amount: float):
    """Process a payment — errors are automatically reported to Sentry."""
    result = payment_gateway.charge(order_id, amount)
    if not result.success:
        raise PaymentError(f"Payment failed: {result.error}")
    return result.transaction_id

When process_payment fails, the error appears in Sentry tagged flexiq.task_name=myapp.tasks.process_payment, flexiq.job_id=..., flexiq.queue=default. Each retry is recorded as a breadcrumb, so the final failure (after all retries) includes the full breadcrumb trail.

import { Queue } from "@byteveda/flexiq";
import * as Sentry from "@sentry/node";
import { sentryMiddleware } from "@byteveda/flexiq/contrib/sentry";

// Initialize Sentry first
Sentry.init({
  dsn: process.env.SENTRY_DSN,
  tracesSampleRate: 1.0,
});

// Create the queue and register the middleware
const queue = new Queue({ dbPath: "myapp.db" });
queue.use(sentryMiddleware());

queue.task(
  "processPayment",
  async (orderId: string, amount: number) => {
    const result = await paymentGateway.charge(orderId, amount);
    if (!result.success) {
      throw new Error(`Payment failed: ${result.error}`);
    }
    return result.transactionId;
  },
  { maxRetries: 3 },
);

If processPayment keeps failing until it exhausts its retries, one event is sent to Sentry — carrying the original stack trace from the last failed attempt plus tags flexiq.task_name=processPayment, flexiq.job_id=..., flexiq.queue=default, flexiq.retry_count=3.

java

import io.sentry.Sentry;
import org.byteveda.flexiq.FlexiQ;
import org.byteveda.flexiq.contrib.SentryMiddleware;
import org.byteveda.flexiq.task.Task;
import org.byteveda.flexiq.worker.Worker;

// Initialize Sentry first
Sentry.init(options -> {
    options.setDsn(System.getenv("SENTRY_DSN"));
    options.setTracesSampleRate(1.0);
});

FlexiQ flexiq = FlexiQ.builder().sqlite("myapp.db").open();
flexiq.use(new SentryMiddleware());

Task<PaymentRequest> processPayment = Task.of("process_payment", PaymentRequest.class)
        .maxRetries(3);

try (Worker worker = flexiq.worker()
        .handle(processPayment, payload ->
                paymentGateway.charge(payload.orderId(), payload.amount()))
        .start()) {
    worker.awaitShutdown();
}

Each failed attempt of process_payment is captured immediately with tags flexiq.task=process_payment, flexiq.job=<job id>. Once retries are exhausted, a final FATAL dead-letter event marks the terminal failure.

Reporting is a line in the task, not a registration on the queue:

rust

use flexiq::{FlexiQ, TaskError};

#[flexiq::task(max_retries = 3, queue = "billing")]
fn process_payment(order_id: String, cents: i64) -> flexiq::Outcome<i64> {
    charge(&order_id, cents).map_err(|err| {
        report(&err); // your own capture call
        TaskError::retryable(err.to_string())
    })?;
    Ok(cents)
}

let queue = FlexiQ::open("myapp.db")?;
let worker = queue.worker().register::<process_payment>().spawn()?;

That reports every failed attempt, because the closure runs on each one. For one event per dead job instead, flexiq_core::Worker::on_outcome delivers ResultOutcome::DeadLettered once retries are spent — but it replaces the worker above rather than adding to it, and it carries the final message rather than the error. The Middleware guide has that trade-off in full.

Combining with other middleware

Where a binding has a middleware stack, the Sentry middleware composes with anything else registered on the same queue or worker — see the Middleware guide for how hooks run across multiple middleware instances.

python

from flexiq.contrib.otel import OpenTelemetryMiddleware
from flexiq.contrib.prometheus import PrometheusMiddleware

queue = Queue(
    db_path="myapp.db",
    middleware=[
        OpenTelemetryMiddleware(),
        PrometheusMiddleware(),
        SentryMiddleware(),
    ],
)

Nothing composes, because there is no stack to compose in. An error report, a metric and a log line all sit wherever you put them; when several want the same failure, give them one small function the task calls on its error path instead of reaching for a hook that is not there.

flexiq_core::Worker::on_outcome will carry exactly one of those concerns — it holds a single callback, and it comes at the price of building the worker on core's pool, which cannot run a task that uses durable steps. Middleware has that trade-off in full.