Stateful Circuit Breaker

The problem

Payment systems can fail or hang, and without a safety switch, these errors can pile up and cause multiple failed transactions. Standard safety switches often forget the history of these errors, making them less accurate at knowing when to stop.

What it does

It monitors payment flows and uses a specific formula to decide when to stop transactions to prevent errors. It also remembers the history of these failures over time to make those decisions more accurate.

Why it matters

It ensures payment integrity by combining smart trip logic with a memory of past system behavior.

Validation

It was run in the sandbox and it failed. run output shows an error/traceback — the artifact does NOT run clean.

$ python3 circuit_breaker.py
Traceback (most recent call last):
  File "/work/circuit_breaker.py", line 64, in <module>
    breaker = CircuitBreaker()
              ^^^^^^^^^^^^^^^^
  File "/work/circuit_breaker.py", line 13, in __init__
    self.failures = self.state.get('failures', 0)
                    ^^^^^^^^^^^^^^
AttributeError: 'str' object has no attribute 'get'

No screenshot — there is nothing working to show. This is recorded as an unfinished sketch so the attempt stays visible instead of being quietly dropped.

The code

All of it — 72 lines, one file, standard library only.

# Stateful Circuit Breaker for Payment Flow
import json
import os
import time
import random

class CircuitBreaker:
    def __init__(self, state_file='circuit_breaker_state.json', threshold=5, cooldown=60):
        self.state_file = state_file
        self.max_failures = threshold
        self.cooldown = cooldown
        self.state = self._load_state().get('state', 'CLOSED')
        self.failures = self.state.get('failures', 0)
        self.last_trip = self.state.get('last_trip', 0)

    def _load_state(self):
        try:
            with open(self.state_file, 'r') as f:
                return json.load(f)
        except (FileNotFoundError, json.JSONDecodeError):
            return {'state': 'CLOSED', 'failures': 0, 'last_trip': 0}

    def _save_state(self):
        state = {
            'state': self.state,
            'failures': self.failures,
            'last_trip': self.last_trip
        }
        with open(self.state_file, 'w') as f:
            json.dump(state, f)

    def _check_state(self):
        now = time.time()
        if self.state == 'OPEN' and now - self.last_trip > self.cooldown:
            self.state = 'HALF_OPEN'
            self._save_state()

    def execute(self, operation):
        self._check_state()
        if self.state == 'OPEN':
            raise Exception('Circuit open - rejecting request')
        try:
            result = operation()
            if self.failures > 0:
                self.failures = 0
                self._save_state()
            return result
        except Exception as e:
            self.failures += 1
            if self.failures >= self.max_failures:
                self.state = 'OPEN'
                self.last_trip = time.time()
                self._save_state()
            raise

# Example usage:
def simulate_payment():
    # Simulate random failures
    if random.random() < 0.3:  # 30% failure rate for demo
        raise Exception('Payment service unavailable')
    return 'Payment successful'

if __name__ == '__main__':
    breaker = CircuitBreaker()
    for i in range(10):
        try:
            print(f'Attempt {i+1}')
            result = breaker.execute(simulate_payment)
            print(result)
        except Exception as e:
            print(f'Error: {str(e)}')
        time.sleep(1)