Streaming 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).

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:

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.

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.

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:

from pyrula.workflows import InMemorySource

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

For Kafka consumption use pyrula.kafka directly, or bind a workflow with @kafka_workflow. Write outputs inside ctx.step.