# Streaming Sources

> Consume an external source inside a workflow and record offsets for replay.

Source: https://docs.pyrula.io/workflows/guides/sources/

`ctx.stream` pulls batches from a `Source` and tracks offsets in the run's journal. On replay it
seeks past committed batches, so the loop body does not run for them. Pass exactly one of these
(neither or both raises `TypeError`):

- `stateless=True`: the loop keeps nothing between batches. Each batch's effects are its own
  steps or writes.
- `carry=<initial value>`: the loop is a fold. Each commit journals its state, which replay
  restores ([below](#folds-carry-their-state)).

```python
from pyrula.workflows import workflow

@workflow(name="ingest")
async def ingest(ctx):
    async for batch in ctx.stream(source, stateless=True, timeout_ms=5000):
        for record in batch.records:
            await handle(record)
        await ctx.commit(source, batch)
```

`ctx.commit` records the batch's next offset. A crash before it repeats the batch, so use
idempotency keys on external effects. Each batch runs in its own replay scope; steps inside it
key per `source:partition:offset`.

## Folds carry their state

A local set while handling a committed batch is not rebuilt after a redrive, because the loop
body does not run for that batch:

```python
total = 0
async for batch in ctx.stream(source, stateless=True):
    total += sum(batch.records)        # wrong: a replay skips committed batches, total restarts at 0
    await ctx.commit(source, batch)
return total
```

Keep it in the carry instead. `ctx.stream` returns a handle: iterate it for the batches, and
read `handle.carry` for the fold's state (the initial value before the first commit, then the
latest commit's carry). `ctx.commit(source, batch, carry=...)` journals the new state in the same
event as the batch's offset. A replay restores the last committed carry, across partitions in
journal order, without re-reading the batches. A batch that crashed before its commit runs again
from the carry before it, so it is not counted twice.

```python
import asyncio

from pyrula.contracts.streams import InMemorySource
from pyrula.workflows import workflow
from pyrula.workflows.testing import WorkflowTestEnv


@workflow(name="order_total")
async def order_total(ctx):
    orders = InMemorySource(source_id="orders", records=[1, 5, 4], batch_size=1)
    total = ctx.stream(orders, carry=0)
    async for batch in total:
        await ctx.commit(orders, batch, carry=total.carry + sum(batch.records))
    return total.carry   # restored from the journal on replay


async def main() -> None:
    run = await WorkflowTestEnv().start(order_total, {})
    assert await run.result() == 10


asyncio.run(main())
```

The carry is stored as JSON, like a step result: a tuple comes back as a list. It rides in the
commit event, so it must fit the workflow's `max_event_payload_bytes`. A carry with no JSON
form raises `TypeError` at `ctx.commit`, one over the cap raises `ValueError`, and neither
commits the batch. A carry that grows with every batch belongs in a store: write it in a step
and carry a key to it.

`ctx.commit` needs `carry=` on a `carry=` stream and refuses it on a `stateless=True` stream, both
with `TypeError`. A run journaled by code without `carry=` and resumed by code that declares it
ends with `run:error` code `stream_carry_missing` once it would skip a commit that recorded no
carry. Finish such runs on the code that started them.

The loop ends when `poll` returns `None` (no data within `timeout_ms`), which completes that
`async for`. A `poll` that raises fails the run.

```python
from pyrula.workflows import Batch   # source_id, partition, offset, records
```

## Sources

Subclass `Source` (`source_id`, `poll`, `seek`, `get_committed`, `partitions`), or use
`InMemorySource` in tests:

```python
from pyrula.workflows import InMemorySource

source = InMemorySource(source_id="orders", records=[...], batch_size=10)
```

For Kafka consumption use [`pyrula.kafka`](https://docs.pyrula.io/kafka.md) directly, or bind a workflow with
[`@kafka_workflow`](https://docs.pyrula.io/workflows/guides/kafka.md). Write outputs inside `ctx.step`.