ObanSentinel.CircuitBreaker (oban_sentinel v0.1.0)

A lightweight ETS-backed circuit breaker for Oban workers.

One process owns the ETS table; telemetry handlers write messages to that process, keeping Oban telemetry callbacks fast. Samples are kept only for the configured rolling window.

Summary

Types

action()

policy()

queue()

snapshot()

status()

worker()

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

record_failure(server, worker, queue)

Records a failed job completion asynchronously.

record_failure(server, worker, queue, duration)

Records a failed job completion and its telemetry duration in native units.

record_start(server, worker, queue)

Records a job start, allowing enqueue/start rates to be inspected from ETS.

record_success(server, worker, queue)

Records a successful job completion asynchronously.

record_success(server, worker, queue, duration)

Records a successful job completion and its telemetry duration in native units.

recover_due(server)

Checks open circuits and attempts recovery after their cooldown.

reset(server, worker, opts \\ [])

Manually closes a worker circuit and optionally resumes its queue.

snapshot(server, worker)

Returns current rolling counters and duration aggregates for a worker.

start_link(opts)

Starts a breaker for one named Oban instance.

status(server, worker)

Returns the current circuit status for a worker.

workers(server)

Lists rolling snapshots for every worker observed by this breaker.

Types

action()

@type action() :: :pause_queue | :emit_only

policy()

@type policy() :: %{
  optional(:failure_threshold) => float(),
  optional(:minimum_samples) => pos_integer(),
  optional(:cooldown_ms) => pos_integer(),
  optional(:action) => action(),
  optional(:auto_recover) => boolean()
}

queue()

@type queue() :: atom() | String.t()

snapshot()

@type snapshot() :: %{
  worker: worker(),
  queue: queue() | nil,
  status: status(),
  samples: non_neg_integer(),
  failures: non_neg_integer(),
  failure_rate: float(),
  starts: non_neg_integer(),
  duration: %{
    count: non_neg_integer(),
    average: float(),
    max: non_neg_integer()
  }
}

status()

@type status() :: :closed | :open

worker()

@type worker() :: module() | String.t()

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

record_failure(server, worker, queue)

@spec record_failure(GenServer.server(), worker(), queue()) :: :ok

Records a failed job completion asynchronously.

record_failure(server, worker, queue, duration)

@spec record_failure(GenServer.server(), worker(), queue(), integer() | nil) :: :ok

Records a failed job completion and its telemetry duration in native units.

record_start(server, worker, queue)

@spec record_start(GenServer.server(), worker(), queue()) :: :ok

Records a job start, allowing enqueue/start rates to be inspected from ETS.

record_success(server, worker, queue)

@spec record_success(GenServer.server(), worker(), queue()) :: :ok

Records a successful job completion asynchronously.

record_success(server, worker, queue, duration)

@spec record_success(GenServer.server(), worker(), queue(), integer() | nil) :: :ok

Records a successful job completion and its telemetry duration in native units.

recover_due(server)

@spec recover_due(GenServer.server()) :: :ok

Checks open circuits and attempts recovery after their cooldown.

reset(server, worker, opts \\ [])

@spec reset(GenServer.server(), worker(), keyword()) :: :ok | {:error, :not_open}

Manually closes a worker circuit and optionally resumes its queue.

snapshot(server, worker)

@spec snapshot(GenServer.server(), worker()) :: snapshot()

Returns current rolling counters and duration aggregates for a worker.

start_link(opts)

@spec start_link(keyword()) :: GenServer.on_start()

Starts a breaker for one named Oban instance.

status(server, worker)

@spec status(GenServer.server(), worker()) :: status()

Returns the current circuit status for a worker.

workers(server)

@spec workers(GenServer.server()) :: [snapshot()]

Lists rolling snapshots for every worker observed by this breaker.