Java/Guides/Reliability/

Circuit Breakers

Trip a task after repeated failures, then sample-probe before fully closing — protect a failing dependency from cascading load.

A circuit breaker stops a task from running once it fails too often within a time window, giving a failing downstream dependency — an external API, a database, a third-party service — time to recover instead of getting hammered by every retry. Enforcement lives entirely in the shared core scheduler; each SDK only supplies the configuration.

Celery has no built-in circuit breaker — this is flexiq-native, configured per task with circuit_breaker={...}.

Coming from BullMQ? BullMQ has no built-in circuit breaker either — it's a flexiq-specific addition. You'd otherwise reach for a separate library (e.g. opossum) alongside a BullMQ Worker.

A breaker is the one resilience knob #[flexiq::task] does not take — the macro rejects circuit_breaker as an unknown attribute. It lives on TaskConfig::circuit_breaker, which means writing a Task impl by hand. The Configuration section below is the whole of it.

How it works

A circuit breaker tracks failures within a time window and transitions through three states:

Closed — normal operation. Tasks execute as usual; each failure is counted, and the count resets once a success comes in or the window expires.
Open — too many failures. Jobs of that task are gated rather than executed.
Half-Open — after the cooldown elapses, a limited number of probe jobs are let through (default 5). Their successes and failures are tracked separately from the closed-state counter. The circuit closes once the success rate among completed probes meets the threshold (default 80%). If enough probes fail that the threshold becomes mathematically impossible before all of them complete, the circuit re-opens immediately. If the probe quota is used up without the cooldown period completing a full cycle, it also re-opens as a safety valve.

A gated job isn't dropped or dead-lettered — the poller reschedules it 5 seconds out and re-checks the breaker the next time it's due, so it runs as soon as the breaker closes (or gets picked as a half-open probe). This 5-second recheck interval is fixed by the core scheduler and isn't configurable from any SDK.

Configuration

A circuit breaker is configured per task, alongside its other resilience settings. It's only enforced once a worker registers that task's config at startup — enqueuing jobs before any such worker has started just queues them normally, since there's nothing yet to gate on.

python

from flexiq import Queue

queue = Queue(db_path="tasks.db")

@queue.task(
    circuit_breaker={
        "threshold": 5,   # open after 5 failures
        "window": 60,     # within a 60-second window
        "cooldown": 120,  # stay open 2 minutes before half-open
    }
)
def call_external_api(endpoint: str) -> dict:
    return requests.get(endpoint).json()

import { Queue } from "@byteveda/flexiq";

const queue = new Queue({ dbPath: "tasks.db" });

queue.task("call_flaky_api", callApi, {
  circuitBreaker: { threshold: 5, windowMs: 60_000, cooldownMs: 120_000 },
});

java

Task<Response> callFlakyApi = Task.of("call_flaky_api", Response.class)
        .circuitBreaker(CircuitBreakerConfig.builder(5)
                .windowSeconds(60)
                .cooldownSeconds(120)
                .build());

circuit_breaker is a field on flexiq::TaskConfig, not a #[flexiq::task] attribute — the macro rejects it. Implement Task over the generated one to set it, reusing everything the macro already worked out:

rust

use flexiq::{CircuitBreakerConfig, EnqueueOptions, Job, Outcome, Task, TaskConfig};

#[flexiq::task]
fn call_external_api(endpoint: String) -> Outcome<String> {
    let body = http::get(&endpoint).map_err(|e| flexiq::TaskError::retryable(e.to_string()))?;
    Ok(body)
}

/// The same task name, with a breaker in front of it.
struct GuardedCall;

impl Task for GuardedCall {
    const NAME: &'static str = <call_external_api as Task>::NAME;

    fn config() -> TaskConfig {
        TaskConfig {
            circuit_breaker: Some(CircuitBreakerConfig {
                threshold: 5,               // open after 5 failures
                window_ms: 60_000,          // within a 60-second window
                cooldown_ms: 120_000,       // stay open 2 minutes before half-open
                half_open_max_probes: 5,
                half_open_success_rate: 0.8,
            }),
            ..<call_external_api as Task>::config()
        }
    }

    fn defaults() -> EnqueueOptions {
        <call_external_api as Task>::defaults()
    }

    fn run_encoded(job: &Job) -> Outcome<Option<Vec<u8>>> {
        <call_external_api as Task>::run_encoded(job)
    }
}

The worker registers GuardedCall; producers keep calling call_external_api::call(..), since both name the same task. CircuitBreakerConfig has no Default, so all five fields are yours to write — 5 and 0.8 are the defaults the core documents.

Parameter reference

Each binding exposes the same breaker with its own names, defaults, and required fields:

Parameter	Type	Default	Description
`threshold`	`int`	`5`	Number of failures to trigger the breaker.
`window`	`int`	`60`	Time window in seconds for counting failures.
`cooldown`	`int`	`300`	Seconds to wait before allowing a half-open probe.
`half_open_probes`	`int`	`5`	Number of probe requests allowed while half-open.
`half_open_success_rate`	`float`	`0.8`	Required success rate (0.0–1.0) among probes to close from half-open.

Field	Type	Default	Description
`threshold`	`number`	required	Failures within `windowMs` that trip the breaker.
`windowMs`	`number`	required	Rolling window for counting failures.
`cooldownMs`	`number`	required	How long the breaker stays open before a half-open probe.
`halfOpenMaxProbes`	`number`	`5`	Probe jobs let through while half-open.
`halfOpenSuccessRate`	`number`	`0.8`	Success rate (0.0–1.0) among probes required to close.

Builder method	Type	Default	Description
`CircuitBreakerConfig.builder(threshold)` / `.of(threshold)`	`int`	required, must be `> 0`	Failure count that trips the breaker.
`.window(Duration)` / `.windowSeconds(long)`	`Duration`	`60s`	Rolling window for counting failures.
`.cooldown(Duration)` / `.cooldownSeconds(long)`	`Duration`	`300s`	How long the breaker stays open before admitting half-open probes.
`.halfOpenProbes(int)`	`int`	`5`	Probe runs admitted while half-open; must be `> 0`.
`.halfOpenSuccessRate(double)`	`double`	`0.8`	Success rate (0.0–1.0) among probes required to re-close; must be within `[0.0, 1.0]`.

CircuitBreakerConfig is a plain struct with five public fields and no Default, so there is no ..Default::default() to lean on — every field is written out, every time.

Field	Type	Default	Description
`threshold`	`i32`	required	Failures within `window_ms` that trip the breaker.
`window_ms`	`i64`	required	Rolling window for counting failures, in milliseconds.
`cooldown_ms`	`i64`	required	How long the breaker stays open before admitting a half-open probe, in milliseconds.
`half_open_max_probes`	`i32`	required	Probe jobs let through while half-open. Core documents `5`.
`half_open_success_rate`	`f64`	required	Success rate (0.0–1.0) among probes required to re-close. Core documents `0.8`.

The whole struct is optional on the task: TaskConfig::circuit_breaker is an Option<CircuitBreakerConfig>, and None — the default — means no breaker at all.

Inspecting circuit breaker state

Only tasks with a registered breaker show up here — plain tasks return nothing.

python

breakers = queue.circuit_breakers()
for cb in breakers:
    print(f"{cb['task_name']}: {cb['state']} (failures: {cb['failure_count']})")

const breakers = await queue.listCircuitBreakers();
for (const cb of breakers) {
  console.log(`${cb.taskName}: ${cb.state} (failures: ${cb.failureCount})`);
}

java

List<CircuitBreakerState> breakers = queue.listCircuitBreakers();
for (CircuitBreakerState cb : breakers) {
    log.info("{}: {} (failures: {})", cb.taskName, cb.state, cb.failureCount);
}

FlexiQ does not wrap the breaker reads; queue.storage() is where they live:

rust

use flexiq::{CircuitState, Storage};

for cb in queue.storage().list_circuit_breakers()? {
    println!(
        "{}: {} (failures: {})",
        cb.task_name,
        CircuitState::from_i32(cb.state).as_str(),
        cb.failure_count
    );
}

CircuitBreakerState::state is the raw i32 discriminant, which is what CircuitState::from_i32 is for.

Dashboard API

These endpoints sit behind the dashboard's normal auth — see each SDK's dashboard guide for setup and authentication details.

bash

curl http://localhost:8080/api/circuit-breakers

json

[
  {
    "task_name": "myapp.tasks.call_external_api",
    "state": "open",
    "failure_count": 5,
    "last_failure_at": 1700000010000,
    "opened_at": 1700000010000,
    "threshold": 5,
    "window_ms": 60000,
    "cooldown_ms": 120000
  }
]

bash

curl http://localhost:8787/api/circuit-breakers

json

[
  {
    "task_name": "call_flaky_api",
    "state": "open",
    "failure_count": 5,
    "last_failure_at": 1700000010000,
    "opened_at": 1700000010000,
    "threshold": 5,
    "window_ms": 60000,
    "cooldown_ms": 120000
  }
]

bash

curl http://localhost:8080/api/circuit-breakers

json

[
  {
    "task_name": "call_flaky_api",
    "state": "open",
    "failure_count": 5,
    "threshold": 5,
    "window_ms": 60000,
    "cooldown_ms": 120000,
    "opened_at": 1700000010000,
    "last_failure_at": 1700000010000,
    "half_open_max_probes": 5,
    "half_open_success_rate": 0.8
  }
]

The Java contract includes half_open_max_probes and half_open_success_rate; the Python and Node contracts don't expose those two fields over the dashboard API today.

The Rust SDK serves no dashboard, so there is no endpoint of its own to call. crates/flexiq-server is what serves the dashboard and this API over a shared backend; in-process, read the same rows with queue.storage().list_circuit_breakers().

When to use

Circuit breakers are most useful for tasks that interact with external systems:

External API calls — prevent hammering a service that's already down
Database connections — stop retrying against an unreachable database
Third-party services — email providers, payment gateways, and similar dependencies

For purely internal computation tasks, circuit breakers are usually unnecessary — plain retries with backoff are sufficient.

Combining with retries

A task can have both. Retries and the breaker are independent counters over the same failures: every failed attempt is recorded against the breaker's threshold — including one that's about to be retried — while the retry budget separately decides whether that attempt gets another try. A sufficiently flaky dependency can trip the breaker mid-retry, before a job's own retry budget is exhausted. Once open, new jobs for that task are gated immediately, so the retry budget stops being spent on a dependency that's already down.

python

@queue.task(
    max_retries=3,
    retry_backoff=2.0,
    circuit_breaker={"threshold": 5, "window": 120, "cooldown": 600},
)
def send_email(to: str, subject: str, body: str):
    smtp.send(to, subject, body)

queue.task("send_email", sendEmail, {
  maxRetries: 3,
  retryBackoff: { baseMs: 2_000, maxMs: 60_000 },
  circuitBreaker: { threshold: 5, windowMs: 120_000, cooldownMs: 600_000 },
});

java

Task<EmailPayload> sendEmail = Task.of("send_email", EmailPayload.class)
        .maxRetries(3)
        .retryPolicy(RetryPolicy.exponential(Duration.ofSeconds(2), Duration.ofMinutes(1)))
        .circuitBreaker(CircuitBreakerConfig.builder(5)
                .windowSeconds(120)
                .cooldownSeconds(600)
                .build());

The retry half is a macro attribute; the breaker half is not, so the two are declared in different places:

rust

#[flexiq::task(max_retries = 3, retry_backoff_ms = "2s", retry_max_delay_ms = "1m")]
fn send_email(to: String, subject: String, body: String) -> flexiq::Outcome<()> {
    smtp::send(&to, &subject, &body).map_err(|e| flexiq::TaskError::retryable(e.to_string()))?;
    Ok(())
}

// ... and on the `Task` impl that carries the breaker:
fn config() -> flexiq::TaskConfig {
    flexiq::TaskConfig {
        circuit_breaker: Some(flexiq::CircuitBreakerConfig {
            threshold: 5,
            window_ms: 120_000,
            cooldown_ms: 600_000,
            half_open_max_probes: 5,
            half_open_success_rate: 0.8,
        }),
        // Keeps max_retries = 3 and the 2-second base delay from the macro.
        ..<send_email as flexiq::Task>::config()
    }
}

See Retries for the backoff curve and dead-letter behavior.

Monitoring state changes

Poll circuit_breakers() / listCircuitBreakers() from a failure hook to alert when a breaker opens. The available hook differs by SDK: Python's JOB_FAILED event fires on every failed attempt, while Node's job.dead and Java's EventName.DEAD only fire once a job's retry budget is exhausted.

python

from flexiq.events import EventType

def monitor_breakers(event_type: EventType, payload: dict) -> None:
    # fires on every failed attempt, including ones that will be retried
    open_breakers = [b for b in queue.circuit_breakers() if b["state"] == "open"]
    if open_breakers:
        names = ", ".join(b["task_name"] for b in open_breakers)
        print(f"WARNING: open circuit breakers: {names}")

queue.on_event(EventType.JOB_FAILED, monitor_breakers)

queue.on("job.dead", async () => {
  // fires once a job's retry budget is exhausted, not on every failed attempt
  const open = (await queue.listCircuitBreakers()).filter((cb) => cb.state === "open");
  if (open.length > 0) {
    console.warn("open circuit breakers:", open.map((cb) => cb.taskName));
  }
});

java

Worker worker = queue.worker()
        .handle(callFlakyApi, payload -> call(payload))
        .on(EventName.DEAD, event -> {
            // fires once a job's retry budget is exhausted, not on every failed attempt
            List<CircuitBreakerState> open = queue.listCircuitBreakers().stream()
                    .filter(CircuitBreakerState::isOpen)
                    .toList();
            if (!open.isEmpty()) {
                log.warn("open circuit breakers: {}", open);
            }
        })
        .start();

There is no failure hook to poll from. WorkerBuilder exposes no outcome callback — flexiq_core::Worker::on_outcome is the nearest seam, and reaching it means building the worker through flexiq_core::Worker directly. Poll on a timer of your own instead:

rust

use flexiq::{CircuitState, Storage};

let open: Vec<String> = queue
    .storage()
    .list_circuit_breakers()?
    .into_iter()
    .filter(|cb| CircuitState::from_i32(cb.state) == CircuitState::Open)
    .map(|cb| cb.task_name)
    .collect();

if !open.is_empty() {
    eprintln!("open circuit breakers: {open:?}");
}