Streaming Error Handling¶

Page Maps¶

graph LR
  family["Python Programming"]
  program["Python Functional Programming"]
  section["Streaming Resilience Failure Handling"]
  page["Streaming Error Handling"]
  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 answers a simple but high-pressure question: once failures are values, how do we keep the stream useful? The answer is not another big control loop. It is a small set of stream combinators with explicit continuation and routing behavior.

Start With the Mixed-Stream Problem¶

Once a pipeline yields Ok and Err values together, you need better tools than "collect everything and sort it out later." One-pass separation and recovery should feel normal here.

If one bad item still kills the whole stream, typed failures have not been integrated properly.
If separating successes from failures requires an early list build, the routing layer is hiding lost laziness.
If recovery or logging changes order or visits items twice, the combinator contract is not clear enough.

Core question:
How do you keep a lazy streaming pipeline flowing when individual records fail, while faithfully collecting every error with full provenance and enabling one-pass routing, recovery, or parallel processing — all without materialising the stream?

This lesson introduces streaming error combinators as the operational layer on top of Result:

keep per-item failures from halting unrelated work
route, recover, or observe results in the same pass that processes them
preserve the original stream order so later folds and reports remain trustworthy

The 99/1 success/failure example matters because it captures the real production pressure: most of the stream is still useful, so the design should not throw that value away.

A naïve pipeline would:

for chunk in chunks:
    embedded.append(embed_chunk(chunk))   # raises → entire pipeline dies

You lose everything after the first bad chunk.

Even a careful try/except loop tends to halt, drop, or prematurely materialize.

The production solution uses small streaming combinators that treat Result as a normal value and keep routing explicit.

embedded = par_try_map_iter(embed_chunk, chunks_with_path, stage="embed")
# → Iterator[Result[Chunk, ErrInfo]] that never raises and preserves order

The stream continues forever; good chunks flow through immediately; every failure is captured with full provenance and can be logged, retried, or routed without stopping anything.

Use this when you run RAG or other data-processing pipelines over real-world messy data and refuse to lose 99 % of the work because of 1 % bad records.

Outcome:
1. You will process mixed good/bad streams with zero halting and O(1) memory per item.
2. You will route, log, recover, or aggregate errors in one pass using lazy combinators.
3. You will ship a RAG pipeline that survives any per-chunk catastrophe and delivers rich, structured error reports.

This section formalises exactly what you should review in this layer: continuation, ordering, one-pass separation, bounded work, and complete containment of wrapped failures.

Concrete Motivating Example¶

Same deep TreeDoc from previous cores:

100 000 chunks total.
99 000 embed successfully.
800 contain truncated UTF-8 → UnicodeDecodeError.
200 are >10 MB → MemoryError in embedder.

Desired behaviour:

embedded: Iterator[Result[Chunk, ErrInfo]] = par_try_map_iter(
    embed_chunk,
    chunks_with_path,
    stage="embed",
    key_path=lambda cp: cp[1],
)

# Never raises, processes all 100k items, yields 99k Ok + 1k Err instantly
oks, errs = partition_results(embedded)   # materialise only at the very end

Total time ≈ time for 99k successful embeddings (the 1k failures are near-instant).
Full provenance on every error (tree path, stage, cause).

1. Laws & Invariants (machine-checked)¶

Law	Formal Statement	Enforcement
Continuation	Pipeline never halts on `Err`; every input item produces exactly one output item.	`test_continuation_full_output`.
Ordering	All combinators preserve input order in their outputs (successes and failures appear in original sequence).	`test_ordering_preservation`.
Separation (lazy)	`filter_ok` / `filter_err` partition the stream into good/bad subsequences, each preserving relative order, without materialisation.	`test_lazy_separation_equivalence`.
Bounded-Work	Processing first k items (good or bad) performs exactly k applications of the wrapped function.	`test_bounded_work`.
Containment	No unhandled exception escapes any combinator; every failure becomes an `Err`.	`test_containment_no_leak`.
One-Pass Routing	`split_results_to_sinks` visits each item exactly once.	`test_single_pass_split`.

These laws guarantee the stream is robust, predictable, and truly lazy.

2. Decision Table – Which Combinator Do You Actually Use?¶

Goal	Need Parallelism?	Need Recovery?	Need One-Pass Routing?	Recommended Combinator
Simple exception containment	No	No	No	`try_map_iter`
Order-preserving parallelism	Yes	No	No	`par_try_map_iter`
Stream only successes	–	No	No	`filter_ok`
Stream/log only failures	–	No	No	`filter_err` / `tap_err`
Recover failures to values	–	Yes	No	`recover_iter`
Recover failures to Result	–	Yes	No	`recover_result_iter`
Route to two sinks in one pass	–	No	Yes	`split_results_to_sinks`
Route + contain sink exceptions	–	No	Yes	`split_results_to_sinks_guarded`
Materialise good/bad at end	–	No	No	`partition_results` (endpoint only)

Never materialise early for separation — use one-pass routing instead.

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

Refactor note: streaming Result combinators live in funcpipe_rag.result.stream (capstone/src/funcpipe_rag/result/stream.py) and are re-exported from funcpipe_rag.result and funcpipe_rag.api.core.

from funcpipe_rag.api.core import (
    filter_err,
    filter_ok,
    par_try_map_iter,
    partition_results,
    recover_iter,
    recover_result_iter,
    split_results_to_sinks,
    split_results_to_sinks_guarded,
    tap_err,
    tap_ok,
    try_map_iter,
)

4. Reference Implementations¶

4.1 try_map_iter – Exception-Safe Lazy Wrapping¶

def try_map_iter(
    fn: Callable[[T], U],
    xs: Iterable[T],
    *,
    stage: str,
    key_path: Callable[[T], tuple[int, ...]] | None = None,
    code: str = "PIPE/EXC",
) -> Iterator[Result[U, ErrInfo]]:
    for x in xs:
        try:
            yield Ok(fn(x))
        except Exception as exc:
            p = key_path(x) if key_path is not None else ()
            yield Err(make_errinfo(code, str(exc), stage, p, exc))

4.2 par_try_map_iter – Order-Preserving Parallel Mapping¶

from collections import deque
from concurrent.futures import ThreadPoolExecutor, Future

def par_try_map_iter(
    fn: Callable[[T], U],
    xs: Iterable[T],
    *,
    stage: str,
    key_path: Callable[[T], tuple[int, ...]] | None = None,
    code: str = "PIPE/EXC",
    max_workers: int = 8,
    max_in_flight: int = 32,
) -> Iterator[Result[U, ErrInfo]]:
    it = iter(xs)
    inflight: deque[tuple[int, T, Future[U]]] = deque()
    idx = 0

    with ThreadPoolExecutor(max_workers=max_workers) as ex:
        # Prime the pipeline
        while len(inflight) < max_in_flight:
            try:
                x = next(it)
            except StopIteration:
                break
            inflight.append((idx, x, ex.submit(fn, x)))
            idx += 1

        out_idx = 0
        while inflight:
            # Drain in order
            while inflight and inflight[0][0] == out_idx:
                i, x, fut = inflight.popleft()
                try:
                    yield Ok(fut.result())
                except Exception as exc:
                    p = key_path(x) if key_path is not None else ()
                    yield Err(make_errinfo(code, str(exc), stage, p, exc))
                out_idx += 1

            # Refill
            if len(inflight) < max_in_flight:
                try:
                    x = next(it)
                except StopIteration:
                    continue
                inflight.append((idx, x, ex.submit(fn, x)))
                idx += 1

4.3 The Rest (concise, correct, lazy)¶

def filter_ok(xs: Iterable[Result[T, E]]) -> Iterator[T]:
    for r in xs:
        if isinstance(r, Ok):
            yield r.value

def filter_err(xs: Iterable[Result[T, E]]) -> Iterator[E]:
    for r in xs:
        if isinstance(r, Err):
            yield r.error

def tap_ok(xs: Iterable[Result[T, E]], fn: Callable[[T], None]) -> Iterator[Result[T, E]]:
    """Observational tap only – fn may log or increment metrics but must not mutate values."""
    for r in xs:
        if isinstance(r, Ok):
            fn(r.value)
        yield r

def tap_err(xs: Iterable[Result[T, E]], fn: Callable[[E], None]) -> Iterator[Result[T, E]]:
    """Observational tap only – fn may log or increment metrics but must not mutate values."""
    for r in xs:
        if isinstance(r, Err):
            fn(r.error)
        yield r

def recover_iter(xs: Iterable[Result[T, E]], fn: Callable[[E], T]) -> Iterator[T]:
    for r in xs:
        yield r.value if isinstance(r, Ok) else fn(r.error)

def recover_result_iter(xs: Iterable[Result[T, E]], fn: Callable[[E], Result[T, E]]) -> Iterator[Result[T, E]]:
    for r in xs:
        yield r if isinstance(r, Ok) else fn(r.error)

def split_results_to_sinks(
    xs: Iterable[Result[T, E]],
    on_ok: Callable[[T], None],
    on_err: Callable[[E], None],
) -> None:
    for r in xs:
        if isinstance(r, Ok):
            on_ok(r.value)
        else:
            on_err(r.error)

def split_results_to_sinks_guarded(
    xs: Iterable[Result[T, E]],
    on_ok: Callable[[T], None],
    on_err: Callable[[E], None],
    *,
    stage: str = "sink",
) -> Iterator[Result[None, ErrInfo]]:
    """Contain sink exceptions; original Err is processed before sink may fail."""
    for r in xs:
        try:
            if isinstance(r, Ok):
                on_ok(r.value)
            else:
                on_err(r.error)
            yield Ok(None)
        except Exception as exc:
            yield Err(make_errinfo("SINK/EXC", str(exc), stage, (), exc))

def partition_results(xs: Iterable[Result[T, E]]) -> tuple[list[T], list[E]]:
    oks: list[T] = []
    errs: list[E] = []
    for r in xs:
        if isinstance(r, Ok):
            oks.append(r.value)
        else:
            errs.append(r.error)
    return oks, errs

4.4 Idiomatic RAG Usage¶

embedded = par_try_map_iter(
    embed_chunk,
    chunks_with_path,
    stage="embed",
    key_path=lambda cp: cp[1],
)

split_results_to_sinks(
    tap_err(embedded, log_err_info),
    on_ok=index_chunk,
    on_err=error_warehouse,
)

5. Property-Based Proofs (`capstone/tests/test_result_stream.py`)¶

@given(items=st.lists(st.integers()))
def test_continuation_full_output(items):
    def f(x: int) -> int:
        if x == -1:
            raise ValueError("boom")
        return x
    results = list(try_map_iter(f, items, stage="test"))
    assert len(results) == len(items)

@given(items=st.lists(st.integers()))
def test_ordering_preservation(items):
    tagged = list(enumerate(items))
    def f(iv):
        i, v = iv
        if v % 2 == 0:
            raise ValueError("even")
        return iv
    results = list(try_map_iter(f, tagged, stage="test", key_path=lambda iv: (iv[0],)))
    ok_indices = [r.value[0] for r in results if isinstance(r, Ok)]
    err_indices = [r.error.path[0] for r in results if isinstance(r, Err)]
    assert ok_indices + err_indices == list(range(len(items)))

@given(items=st.lists(st.integers()))
def test_lazy_separation_equivalence(items):
    def f(x: int) -> int:
        if x % 2 == 0:
            raise ValueError("even")
        return x
    stream = try_map_iter(f, items, stage="test")
    oks = list(filter_ok(stream))
    stream2 = try_map_iter(f, items, stage="test")
    errs = list(filter_err(stream2))
    assert len(oks) + len(errs) == len(items)

@given(items=st.lists(st.integers()))
def test_single_pass_split(items):
    seen = 0
    def on_ok(_): nonlocal seen; seen += 1
    def on_err(_): nonlocal seen; seen += 1
    def f(x: int) -> int:
        if x % 2 == 0:
            raise ValueError("even")
        return x
    split_results_to_sinks(try_map_iter(f, items, stage="test"), on_ok, on_err)
    assert seen == len(items)

@given(items=st.lists(st.integers()))
def test_bounded_work(items):
    seen = 0
    def f(x):
        nonlocal seen
        seen += 1
        if x == 0:
            raise ValueError("zero")
        return x
    stream = try_map_iter(f, items, stage="test")
    list(islice(stream, 25))
    assert seen == min(25, len(items))

@given(items=st.lists(st.integers()))
def test_try_map_iter_matches_try_except(items):
    def f(x: int) -> int:
        if x == 0:
            raise ValueError("boom")
        return 100 // x

    # reference try/except
    ref: list[Result[int, ErrInfo]] = []
    for x in items:
        try:
            ref.append(Ok(100 // x))
        except Exception as exc:
            # match production arity: include `exc` as cause
            ref.append(Err(make_errinfo("PIPE/EXC", str(exc), "test", (), exc)))

    got = list(try_map_iter(f, items, stage="test"))

    # Compare shapes, values, and error codes/messages (cause/path may differ)
    assert [isinstance(r, Ok) for r in got] == [isinstance(r, Ok) for r in ref]
    assert [r.value for r in got if isinstance(r, Ok)] == \
           [r.value for r in ref if isinstance(r, Ok)]
    assert [r.error.code for r in got if isinstance(r, Err)] == \
           [r.error.code for r in ref if isinstance(r, Err)]

@given(items=st.lists(st.integers(), min_size=1, max_size=200))
def test_par_try_map_iter_matches_try_map_iter(items):
    def f(x: int) -> int:
        if x == 0:
            raise ValueError("boom")
        return 100 // x

    seq = list(try_map_iter(f, items, stage="test"))
    par = list(par_try_map_iter(f, items, stage="test", max_workers=4, max_in_flight=8))

    assert [isinstance(r, Ok) for r in par] == [isinstance(r, Ok) for r in seq]
    assert [r.value if isinstance(r, Ok) else r.error.msg for r in par] == \
           [r.value if isinstance(r, Ok) else r.error.msg for r in seq]

@given(items=st.lists(st.integers()))
def test_containment_no_leak(items):
    def f(x: int) -> int:
        raise ValueError("always fail")
    # This should never raise
    list(try_map_iter(f, items, stage="test"))
    list(par_try_map_iter(f, items, stage="test"))

6. Big-O & Allocation Guarantees¶

Variant	Time per item	Heap per item	Laziness
try_map_iter	O(1)	O(1)	Yes
par_try_map_iter	O(1) amortised	O(max_in_flight) total	Yes
filter_ok / filter_err / tap_*	O(1)	O(1)	Yes
recover_iter / recover_result_iter	O(1)	O(1)	Yes
split_results_to_sinks	O(1)	O(1)	Yes
partition_results	O(1)	O(N) total	No

The serial combinators request one item at a time. par_try_map_iter keeps up to max_in_flight submitted futures, so it is bounded but may request ahead of the next emitted item. “Lazy” does not imply identical demand for serial and parallel routes.

7. Anti-Patterns & Immediate Fixes¶

Anti-Pattern	Symptom	Fix
Halting on first exception	Lost data after first failure	Use `try_map_iter` / `par_try_map_iter`
Materialising early for separation	Memory blowup	Use `split_results_to_sinks` (one-pass)
Silent drop of bad records	Incomplete results	Use `filter_err` or `tap_err` to capture
Sink exceptions crashing pipeline	Partial processing	Use `split_results_to_sinks_guarded`

Trace one failure without losing its stream position¶

The Module 03 pipeline already preserves document and chunk order. Module 04 must preserve that useful observation when an embedding call fails. Returning only successful values would shift every later position; raising would erase all later records. try_map_iter instead emits exactly one Result for each input it requests.

The focused learning source is:

(0, "first"), (1, "bad"), (2, "unrequested")

Taking two outputs produces:

Requested input	Output	Evidence retained
`(0, "first")`	`Ok("FIRST")`	Successful value
`(1, "bad")`	`Err(EMBED/INVALID)`	Stage `embed`, path `(1,)`, original cause
`(2, "unrequested")`	no output	Source item is not requested

Run:

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 result_stream

The requested == 2 assertion is as important as the two output assertions. It connects failure handling back to Module 03's bounded-demand contract.

Choose the next combinator based on what the consumer needs:

Consumer decision	Route	Information or cost
Continue with success and failure in encounter order	Keep `Iterator[Result[...]]`	No early materialization
Observe successes without changing the stream	`tap_ok`	Side effect occurs during consumption
Observe failures without changing the stream	`tap_err`	Side effect can itself fail unless guarded
Route values to two final sinks	`split_results_to_sinks`	Terminal consumption; no return stream
Recover a known failure to a value	`recover_iter`	Original failure no longer appears downstream
Build two complete lists	`partition_results`	O(N) storage and finite-input requirement

try_map_iter intentionally catches Exception at a generic adapter boundary. That is broad. Do not infer that every stage should catch every exception. Where the domain knows the expected failures, translate those specific exceptions into stable error codes and let unexpected defects surface.

The parallel route preserves output order but can have several calls already running when a consumer stops. Its max_in_flight bound controls submitted work, not instantaneous cancellation of work already submitted.

8. Pre-Core Quiz¶

Why emit one Result per requested input? → To retain position while allowing later records to continue.
What does islice(..., 2) prove on the serial route? → Only two source items were requested.
Does par_try_map_iter have the same demand? → No; it may submit up to its in-flight bound.
When does recovery lose evidence? → When an Err is converted to a fallback without separately retaining the failure.
Why is broad exception capture confined to an adapter? → The adapter has an explicit containment contract; domain functions should classify more narrowly.

9. Post-Core Exercise¶

Extend the learning source with one permanent and one transient embedding failure.

Keep one output per requested input.
Assign distinct codes while preserving the same embed stage.
Use paths to recover original positions without inspecting messages.
Take only the first three outputs and assert the fourth input is not requested.
Explain how the demand assertion would change for par_try_map_iter with max_in_flight=2.

Move forward when you can review values, ordering, provenance, and demand as four distinct claims.

Continue with: Error Aggregation

The stream now retains both successes and failures. The next core decides how much of that stream must be consumed to produce an aggregate decision.