Resilience Exceptions

Exceptions for the resilience module: circuit breaker, graceful shutdown, and heartbeat.

Exception hierarchy

RuntimeError
├── CircuitBreakerError
│   └── CircuitOpenError      # Circuit is open, call blocked
├── HeartbeatError            # Heartbeat file write/read failure
└── ShutdownError             # Shutdown callback failure

Common failure modes

• CircuitOpenError is raised when calling through an open circuit. Contains remaining_seconds until half-open transition.

• HeartbeatError surfaces when the state file cannot be written (permissions, disk full) or read (corrupted JSON).

• ShutdownError may be raised if a cleanup callback fails catastrophically (though errors are normally caught).

Usage patterns

Handling circuit breaker fast-fail

from kstlib.resilience import CircuitBreaker, CircuitOpenError

cb = CircuitBreaker(max_failures=3, reset_timeout=30)

try:
    result = cb.call(fetch_data, symbol="BTC/USDT")
except CircuitOpenError as e:
    # Circuit is open - use fallback
    logger.warning(f"Circuit open, retry in {e.remaining_seconds:.1f}s")
    result = get_cached_fallback()

Monitoring heartbeat failures

from kstlib.resilience import Heartbeat, HeartbeatError

def on_missed_beat(error: Exception) -> None:
    logger.error(f"Heartbeat write failed: {error}")
    alerting.notify("Heartbeat failure detected")

heartbeat = Heartbeat(
    state_file="/tmp/bot.heartbeat",
    interval=10,
    on_missed_beat=on_missed_beat,
)

API reference

Specialized exceptions raised by the kstlib.resilience module.

exception kstlib.resilience.exceptions.CircuitBreakerError

Bases: KstlibError, RuntimeError

Base exception for circuit breaker errors.

exception kstlib.resilience.exceptions.CircuitOpenError(message, remaining_seconds)

Bases: CircuitBreakerError

Raised when a call is attempted while the circuit is open.

remaining_seconds

Time until the circuit may transition to half-open.

__init__(self, message: 'str', remaining_seconds: 'float') → 'None' -> None

Initialize CircuitOpenError.

Parameters:

• message (str) – Human-readable error message.

• remaining_seconds (float) – Seconds until circuit may transition to half-open.

exception kstlib.resilience.exceptions.HeartbeatError

Bases: KstlibError, RuntimeError

Raised when the heartbeat encounters an error.

Examples include state file write failure or invalid state file path.

exception kstlib.resilience.exceptions.RateLimitError

Bases: KstlibError, RuntimeError

Base exception for rate limiter errors.

exception kstlib.resilience.exceptions.RateLimitExceededError(message, retry_after)

Bases: RateLimitError

Raised when rate limit is exceeded and blocking is disabled.

retry_after

Seconds until a token will be available.

__init__(self, message: 'str', retry_after: 'float') → 'None' -> None

Initialize RateLimitExceededError.

Parameters:

• message (str) – Human-readable error message.

• retry_after (float) – Seconds until a token will be available.

exception kstlib.resilience.exceptions.ShutdownError

Bases: KstlibError, RuntimeError

Raised when graceful shutdown encounters an error.

Examples include cleanup callback failure or timeout exceeded.

exception kstlib.resilience.exceptions.WatchdogError

Bases: KstlibError, RuntimeError

Base exception for watchdog errors.

exception kstlib.resilience.exceptions.WatchdogTimeoutError(message, seconds_inactive)

Bases: WatchdogError

Raised when watchdog detects inactivity timeout.

seconds_inactive

Time since last ping/activity.

__init__(self, message: 'str', seconds_inactive: 'float') → 'None' -> None

Initialize WatchdogTimeoutError.

Parameters:

• message (str) – Human-readable error message.

• seconds_inactive (float) – Seconds since last activity.