partition_loop

partition_loop is the standard transform mode: compiled feed supplies work, domain processor returns logical results, and framework persists them.

Lifecycle

1. compile manifest and merged params;
2. create StorageSession and metrics;
3. compile StepFeedPlan and enumerate WorkUnits;
4. load/resume registry state;
5. initialize processor context;
6. materialize WorkUnits as ProcessorBatch;
7. call batch_processor on CPU batches;
8. merge ProcessResult rows and write lanes;
9. checkpoint registry/output state according to DRTML;
10. finalize manifests, metrics and result metadata.

Exact processor contract

def process_batch(
    raws: list[Any],
    *,
    ctx: RunContext,
    params: Mapping[str, Any],
    processor_ctx: Mapping[str, Any],
    partition_id: str,
) -> list[ProcessResult]:
    ...

Each item is normally ProcessorBatch(partition_id, source_run_id, primary_rows, attach, shard_id, work_unit_id). The result uses:

ProcessResult(
    ok=True,
    rows={"logical_output": [{"id": "1", "value": "x"}]},
    stats={"items_total": 1},
    shard_id="source-shard",
)

processing.outputs maps logical output names to dataset ids. The processor never calls storage.

Processor context

Optional processing.init is called as init(ctx=ctx, params=params, resume=resume) and must return a dict. Framework may add prefetched mapping indexes or embedding inference state. Context is in-memory reusable state, not a lifecycle escape hatch.

Registries

Framework loads a declared registry from dataset rows on resume, lets domain processing update in-memory state, drains rows through registry_flush, and applies checkpoint cadence from output dataset declarations.

Profile and params

PartitionRunProfile reads generic runtime.* settings such as batch limits, progress logging, resume and metrics. It never interprets domain.*.

Parallel backends

runtime.parallelism.backend selects inprocess, processes, or ray. Multi-worker dispatch assigns primary shards and creates independent worker sessions. Checkpoint/resume and run-global mutable registry semantics currently require inprocess execution.

Module boundaries

• storage/feed: WorkUnit enumeration/materialization.
• storage/run: writes, registry load/drain and checkpoint policy.
• processing/partition_loop: lifecycle glue.
• processing/partition_loop/parallel: worker dispatch/transport.
• telemetry: Prometheus emission.