Skip to content

Circuit Breakers

Page Maps

graph LR
  family["Python Programming"]
  program["Python Functional Programming"]
  section["Streaming Resilience Failure Handling"]
  page["Circuit Breakers"]
  capstone["Capstone evidence"]

  family --> program --> section --> page
  page -.applies in.-> capstone
flowchart LR
  orient["Orient on the page map"] --> read["Read the main claim and examples"]
  read --> inspect["Inspect the related code, proof, or capstone surface"]
  inspect --> verify["Run or review the verification path"]
  verify --> apply["Apply the idea back to the module and capstone"]

This lesson makes circuit breakers feel like an act of design honesty, not panic. A breaker is justified when continuing no longer creates useful value, and the stopping rule itself must be explicit enough to review.

Start With the Hopeless Run

By this point in the module, the stream can survive bad records. The new question is when survival stops being the right goal and early termination becomes the responsible one.

  • If error rate or error count has already crossed a threshold that makes the run useless, continuing is wasted work.
  • If the stopping rule is hidden in loop state, reviewers cannot tell why the run stopped when it did.
  • If early termination does not trigger cleanup promptly, the breaker has solved one problem while causing another.

Core question:
How do you implement short-circuiting and circuit-breaker patterns in streaming pipelines using pure Result types, ensuring early termination on thresholds or failures while maintaining purity, composability, and resource safety?

This lesson introduces breakers as explicit stop policies over Result streams:

  • define the threshold in data rather than hiding it inside ad hoc loop logic
  • stop as soon as the policy says the run has become hopeless
  • preserve composability by emitting or truncating in a predictable, reviewable way

The motivating error-rate example matters because it captures the real tradeoff clearly: there is a point where continuing the run no longer buys meaningful information.

The naïve solution is a manual flag inside a loop:

error_rate = 0.0
seen = 0
for r in embedded:
    seen += 1
    if isinstance(r, Err):
        n_err += 1
    if seen >= 500:
        error_rate = n_err / seen
        if error_rate > 0.2:
            logger.critical("Aborting run – error rate too high")
            break
    process(r)

This works once — but it’s duplicated everywhere, easy to get wrong, and breaks when you later add parallelism or recovery.

The production solution expresses that stopping rule as a lazy, composable breaker over the stream, so the threshold and resulting behavior are both visible.

Use this when you run long-running batch pipelines and cannot afford to process doomed data for hours.

Outcome:
1. You will short-circuit on first error, error count, error rate, or arbitrary predicate — all with O(k) work.
2. You will choose between observable breakers (emit BreakInfo) and silent truncate breakers.
3. You will ship a RAG pipeline that aborts gracefully the moment it becomes hopeless, with full provenance on why.

This section formalises exactly what you should review in breaker code: short-circuiting position, preserved ordering up to the break, bounded work, cleanup behavior, and equivalence to a straightforward reference policy.

Concrete Motivating Example

Same 100 000 chunk tree from previous cores:

  • Normal success rate ≈ 99 %.
  • One malformed section causes 5 000 consecutive failures → error rate spikes to 30 % after 10 000 chunks processed.

Desired behaviour:

embedded = circuit_breaker_rate_emit(
    embedded,
    max_rate=0.2,
    min_samples=500,
)

for r in embedded:
    if isinstance(r, Err) and isinstance(r.error, BreakInfo):
        report_circuit_break(r.error)
        break
    if isinstance(r, Ok):
        index_chunk(r.value)
    else:
        log_err_info(r.error)

Total work: ~10 500 chunk attempts (stops instantly when threshold hit).

1. Laws & Invariants (machine-checked)

Law Formal Statement Enforcement
Short-Circuit Emit/truncate breakers stop after trigger, performing O(k) work where k ≤ position of trigger. test_emit_breakers_short_circuit, test_truncate_breakers_stop_silently.
Ordering Items (including terminal BreakInfo) appear in original stream order. test_breaker_ordering.
Resource Cleanup Upstream generators are closed via their close() method on early termination. test_upstream_closed_on_break.
Purity Breakers are pure functions — deterministic, no side effects beyond iteration. Reproducibility tests.
Equivalence For finite streams, breaker output equals the full stream truncated at the first trigger position, optionally followed by a terminal BreakInfo. test_breaker_equivalence_to_full_scan.

These laws guarantee breakers are safe, predictable, and resource-correct.

2. Decision Table – Which Breaker Do You Actually Use?

Trigger Need Observable Break? Need Terminal Value? Recommended Variant
First error Yes Yes short_circuit_on_err_emit
First error (silent) No No short_circuit_on_err_truncate
Error rate threshold Yes Yes circuit_breaker_rate_emit
Error rate threshold (silent) No No circuit_breaker_rate_truncate
Error count threshold Yes Yes circuit_breaker_count_emit
Custom predicate Yes Yes circuit_breaker_pred_emit

Prefer emit variants — they give you a terminal BreakInfo for reporting without breaking composability.

3. Public API Surface (end-of-Module-04 refactor note)

Refactor note: breakers live in funcpipe_rag.policies.breakers (capstone/src/funcpipe_rag/policies/breakers.py) and are re-exported from funcpipe_rag.api.core.

All snippets assume the Result / Ok / Err ADT from earlier Module 4 cores is already imported.

from funcpipe_rag.api.core import (
    BreakInfo,
    circuit_breaker_count_emit,
    circuit_breaker_count_truncate,
    circuit_breaker_pred_emit,
    circuit_breaker_pred_truncate,
    circuit_breaker_rate_emit,
    circuit_breaker_rate_truncate,
    short_circuit_on_err_emit,
    short_circuit_on_err_truncate,
)

4. Reference Implementations

All breakers are pure generators and guarantee upstream cleanup via try/finally.

4.1 Emit Breakers (Observable Termination)

def short_circuit_on_err_emit(
    xs: Iterable[Result[T, E]],
) -> Iterator[Result[T, E | BreakInfo[E]]]:
    """Yield items until first Err (which is yielded), then emit terminal BreakInfo."""
    it = iter(xs)
    n_ok = n_err = 0
    last_err: E | None = None
    exhausted = False
    try:
        for r in it:
            yield r
            if isinstance(r, Ok):
                n_ok += 1
            else:
                n_err += 1
                last_err = r.error
                bi = BreakInfo(
                    code="BREAK/FIRST_ERR",
                    reason="first error encountered",
                    last_error=last_err,
                    n_ok=n_ok,
                    n_err=n_err,
                    total=n_ok + n_err,
                    threshold=MappingProxyType({}),
                )
                yield Err(bi)
                return
        exhausted = True
    finally:
        if not exhausted:
            close = getattr(it, "close", None)
            if callable(close):
                close()

def circuit_breaker_rate_emit(
    xs: Iterable[Result[T, E]],
    *,
    max_rate: float,
    min_samples: int = 100,
) -> Iterator[Result[T, E | BreakInfo[E]]]:
    """Yield until error rate > max_rate after min_samples, then emit terminal BreakInfo."""
    if not 0.0 < max_rate < 1.0:
        raise ValueError("max_rate must be in (0,1)")
    if min_samples < 1:
        raise ValueError("min_samples >= 1")
    it = iter(xs)
    n_ok = n_err = 0
    last_err: E | None = None
    exhausted = False
    try:
        for r in it:
            yield r
            if isinstance(r, Ok):
                n_ok += 1
            else:
                n_err += 1
                last_err = r.error
            total = n_ok + n_err
            if total >= min_samples and n_err / total > max_rate:
                bi = BreakInfo(
                    code="BREAK/ERR_RATE",
                    reason=f"error rate {n_err/total:.3f} > {max_rate}",
                    last_error=last_err,
                    n_ok=n_ok,
                    n_err=n_err,
                    total=total,
                    threshold=MappingProxyType({"max_rate": max_rate, "min_samples": min_samples}),
                )
                yield Err(bi)
                return
        exhausted = True
    finally:
        if not exhausted:
            close = getattr(it, "close", None)
            if callable(close):
                close()

def circuit_breaker_count_emit(
    xs: Iterable[Result[T, E]],
    *,
    max_errs: int,
) -> Iterator[Result[T, E | BreakInfo[E]]]:
    """Yield until error count > max_errs, then emit terminal BreakInfo."""
    if max_errs < 0:
        raise ValueError("max_errs >= 0")
    it = iter(xs)
    n_ok = n_err = 0
    last_err: E | None = None
    exhausted = False
    try:
        for r in it:
            yield r
            if isinstance(r, Err):
                n_err += 1
                last_err = r.error
                if n_err > max_errs:
                    bi = BreakInfo(
                        code="BREAK/ERR_COUNT",
                        reason=f"errors {n_err} > {max_errs}",
                        last_error=last_err,
                        n_ok=n_ok,
                        n_err=n_err,
                        total=n_ok + n_err,
                        threshold=MappingProxyType({"max_errs": max_errs}),
                    )
                    yield Err(bi)
                    return
            else:
                n_ok += 1
        exhausted = True
    finally:
        if not exhausted:
            close = getattr(it, "close", None)
            if callable(close):
                close()

def circuit_breaker_pred_emit(
    xs: Iterable[Result[T, E]],
    pred: Callable[[Result[T, E]], bool],
) -> Iterator[Result[T, E | BreakInfo[E]]]:
    """Yield until pred(r) is True, then emit terminal BreakInfo."""
    it = iter(xs)
    n_ok = n_err = 0
    last_err: E | None = None
    exhausted = False
    try:
        for r in it:
            yield r
            if isinstance(r, Ok):
                n_ok += 1
            else:
                n_err += 1
                last_err = r.error
            if pred(r):
                bi = BreakInfo(
                    code="BREAK/PRED",
                    reason="predicate triggered",
                    last_error=last_err,
                    n_ok=n_ok,
                    n_err=n_err,
                    total=n_ok + n_err,
                    threshold=MappingProxyType({}),
                )
                yield Err(bi)
                return
        exhausted = True
    finally:
        if not exhausted:
            close = getattr(it, "close", None)
            if callable(close):
                close()

4.2 Truncate Breakers (Silent Termination)

def short_circuit_on_err_truncate(xs: Iterable[Result[T, E]]) -> Iterator[Result[T, E]]:
    """Yield until first Err, then stop silently. No terminal value."""
    it = iter(xs)
    exhausted = False
    try:
        for r in it:
            yield r
            if isinstance(r, Err):
                return
        exhausted = True
    finally:
        if not exhausted:
            close = getattr(it, "close", None)
            if callable(close):
                close()

(The other truncate variants follow the same pattern — return instead of yielding BreakInfo.)

4.3 Idiomatic RAG Usage

embedded = circuit_breaker_rate_emit(
    embedded,
    max_rate=0.2,
    min_samples=500,
)

for r in embedded:
    if isinstance(r, Err) and isinstance(r.error, BreakInfo):
        report_circuit_break(r.error)
        break
    if isinstance(r, Ok):
        index_chunk(r.value)
    else:
        log_err_info(r.error)

5. Property-Based Proofs (capstone/tests/test_breakers.py)

@given(items=st.lists(st.integers()).filter(lambda v: 0 in v))
def test_emit_breakers_short_circuit(items):
    first_err_pos = items.index(0)
    def f(x: int) -> Result[int, str]:
        return Ok(x) if x != 0 else Err("ZERO")
    results = list(short_circuit_on_err_emit(map_result_iter(f, items)))
    assert len(results) == first_err_pos + 2  # items + Err + BreakInfo
    assert isinstance(results[-1], Err) and isinstance(results[-1].error, BreakInfo)
    bi = results[-1].error
    assert bi.n_ok == first_err_pos
    assert bi.n_err == 1
    assert bi.total == bi.n_ok + bi.n_err
    assert bi.last_error == "ZERO"

@given(items=st.lists(st.integers()))
def test_truncate_breakers_stop_silently(items):
    def f(x: int) -> Result[int, str]:
        return Ok(x) if x != 0 else Err("ZERO")
    results = list(short_circuit_on_err_truncate(map_result_iter(f, items)))
    if 0 in items:
        assert len(results) == items.index(0) + 1
    else:
        assert len(results) == len(items)

@given(items=st.lists(st.integers()))
def test_upstream_closed_on_break(items):
    closed = False
    sentinel_seen = False
    def src():
        nonlocal closed, sentinel_seen
        try:
            for x in items:
                yield Ok(x) if x != 0 else Err("ZERO")
                if x == 0:
                    yield Ok("should not be reached")
                    sentinel_seen = True
        finally:
            closed = True
    results = list(short_circuit_on_err_truncate(src()))
    assert not sentinel_seen
    assert closed

@given(items=st.lists(st.integers()))
def test_breaker_ordering(items):
    def f(x: int) -> Result[int, str]:
        return Ok(x) if x != 0 else Err("ZERO")
    full = list(map_result_iter(f, items))
    broken = list(short_circuit_on_err_emit(map_result_iter(f, items)))
    # Strip terminal BreakInfo if present
    if broken and isinstance(broken[-1], Err) and isinstance(broken[-1].error, BreakInfo):
        prefix = broken[:-1]
    else:
        prefix = broken
    assert prefix == full[:len(prefix)]

@given(items=st.lists(st.integers()))
def test_breaker_equivalence_to_full_scan(items):
    def f(x: int) -> Result[int, str]:
        return Ok(x) if x != 0 else Err("ZERO")
    full = list(map_result_iter(f, items))
    broken = list(short_circuit_on_err_emit(map_result_iter(f, items)))
    # Strip terminal BreakInfo if present
    if broken and isinstance(broken[-1], Err) and isinstance(broken[-1].error, BreakInfo):
        prefix = broken[:-1]
    else:
        prefix = broken
    assert prefix == full[:len(prefix)]

def test_count_breaker_off_by_one():
    xs: list[Result[int, str]] = [Err("E"), Err("E"), Err("E")]
    results = list(circuit_breaker_count_emit(xs, max_errs=1))
    # first tolerated Err + threshold-crossing Err + terminal BreakInfo
    assert len(results) == 3
    assert isinstance(results[0], Err)
    assert isinstance(results[2], Err) and isinstance(results[2].error, BreakInfo)
    bi = results[2].error
    assert bi.n_err == 2
    assert bi.threshold["max_errs"] == 1

6. Big-O & Allocation Guarantees

Variant Time Heap Laziness
*_emit breakers O(k) on trigger / O(N) O(1) Yes
*_truncate breakers O(k) on trigger / O(N) O(1) Yes

All breakers are truly lazy generators with O(1) auxiliary memory.

7. Anti-Patterns & Immediate Fixes

Anti-Pattern Symptom Fix
Manual flag-based early exit Duplicated buggy code Use *_emit or *_truncate breakers
Continuing after fatal error rate Wasted hours of compute Use circuit_breaker_rate_*
Resource leaks on early break Open files/connections All breakers close upstream on termination

Read the terminal value as part of the stream protocol

A breaker answers a different question from a fail-fast fold. A fold returns one aggregate result. An emitting breaker preserves the consumed prefix and adds a terminal Err(BreakInfo(...)) explaining why demand stopped.

For short_circuit_on_err_emit:

upstream:  Ok(1), Err("unavailable"), Ok(2)
downstream: Ok(1), Err("unavailable"), Err(BreakInfo), <stop>

The threshold-crossing error remains visible. BreakInfo is an additional control observation, not a replacement for that record failure.

For count breakers, max_errs means “how many errors are tolerated.” With max_errs=1, the first error is tolerated and the second triggers:

Emission n_err after emission Decision
first Err 1 continue
second Err 2 emit it, then emit terminal BreakInfo
third input — never requested

That is why the unit test has three outputs: two source errors and one terminal value.

Run the focused application proof:

course=programs/python-programming/python-functional-programming
state="$course/capstone/module-reference-states/module-04"
venv=artifacts/venv/python-programming/python-functional-programming/capstone

PYTHONPATH="$state/src" \
  "$venv/bin/pytest" -q \
  "$state/tests/learning/test_module_04_resilient_streaming.py" \
  -k emitting_breaker

The source sets closed=True in finally. The breaker calls close() on its upstream iterator when it returns early, so the assertion proves this particular generator's cleanup ran. It does not prove that arbitrary upstream resources close themselves; the next core supplies an explicit lifetime boundary.

Choose emit versus truncate deliberately:

Contract Use
A report, CLI, or operator must know why the run stopped Emitting breaker
The caller already owns termination evidence and requires only the prefix Truncating breaker
Silent termination could be mistaken for normal exhaustion Do not truncate

Rate policies also need a meaningful sample floor. A single failure out of one item has a 100% observed rate but usually too little evidence for a run-level health judgment. min_samples makes that uncertainty explicit; it does not make the chosen threshold statistically valid by itself.

8. Pre-Core Quiz

  1. With max_errs=1, which error triggers? → The second error.
  2. Is the triggering record preserved by an emitting breaker? → Yes, followed by a terminal BreakInfo.
  3. Why can truncation be dangerous? → The prefix can look like normal exhaustion when no separate evidence exists.
  4. What does upstream close() prove? → That a cooperative closable iterator was asked to terminate.
  5. Why set min_samples on a rate breaker? → To avoid making a run-level decision from an arbitrarily tiny prefix.

9. Post-Core Exercise

Add a count breaker to a mixed embedding stream.

  • Predict every emitted value for max_errs=0 and max_errs=1.
  • Assert that the threshold-crossing failure appears before BreakInfo.
  • Count source requests and prove the item after the trigger is not requested.
  • Compare emitting and truncating outputs and state which is acceptable for an operator-facing run.
  • Keep cleanup as a separate assertion rather than inferring it from output length.

Move forward when you can explain the difference between a record failure, a terminal policy event, and normal exhaustion.

Continue with: Resource-Aware Streams

Early termination is now observable. The next core makes the lifetime of the source explicit even when consumers, producers, or policies stop the stream.