Sentry Integration

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

flexiq ships a Sentry middleware that 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 — call Sentry.init(...) in your own app before registering the middleware.

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")

Initialize Sentry, then register the middleware on the queue:

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

None of the three bindings call Sentry.init for you — configure the DSN, environment, and sampling with the Sentry SDK itself. When Sentry hasn't been initialized, the middleware's hooks are safe no-ops.

What it captures

Each binding decides when to report a failure differently:

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

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."));

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.

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.

Combining with other middleware

Sentry middleware composes with any other middleware 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(),
    ],
)