# orca.pipeline

> A module of morca's API.

Pipelines: a named list of steps, checked and run in order.

```mojo
comptime Taxi = Pipeline["taxi", Download, Load, Compact]

var result = Taxi.run(parse_months("2024-01..2024-03"))
```

A pipeline is partitioned by month unless it says otherwise with
`partitions=` (`orca.partitions`): its first step takes the partition key,
or a `Seq` of them for a gather step.

Steps are listed after the step they follow, so the declared order is a
valid run order; `check` asserts it at compile time, along with every edge's
types and unique names. Every function that runs a pipeline calls `check`,
so a pipeline that breaks a rule doesn't compile.

`execute` is the executor: it runs the steps that haven't succeeded yet and
reports each attempt to a `Store`, either `MemoryStore` (what `run` uses: a
run lives as long as the process) or `orca.state.PgStore`, which makes a run
resumable by another process. Each step's output is kept encoded by dista's codec, as Postgres
holds it. A step that raises is retried up to its `retries`, then fails its
partition: the steps after it for that partition, and any gather step over
it, are skipped, while other partitions carry on. One that raises
`orca.step.defer` ends `deferred`, with no more retries, and its downstream
steps are skipped likewise; the run is `deferred` if no step failed.

## Structs

### `Pipeline`

```mojo
struct Pipeline[name: StringSpan[ImmStaticOrigin], *Ts: StepLike, *, tasks: TaskSet = NoTasks, partitions: Partitions = Monthly, schedule: ScheduleLike = NoSchedule, fills: Fill = Previous, catchup: Bool = True, max_runs: Int = Int(1), run_retries: Int = Int(3), retry_seconds: Int = Int(0), max_retry_seconds: Int = Int(0), sensor: Bool = False, keys_per_run: Int = Int(1), workers: Int = Int(2)]
```

A pipeline named `name` of the steps `Ts`, in run order, whose steps call the dista `tasks`, over the keys of `partitions`; run by `schedule` for the partition its fires `fills` (`orca.schedule`), and, with `sensor`, when a dataset it reads gets a new snapshot (`orca.sensors`). A backfill puts `keys_per_run` keys in each run it queues, and so do a fill and, above 1, a sensor (#240; orca.toml's `keys_per_run` overrides it): one for time keys by default, tens to hundreds for keyed partitions (`orca.keyed`), whose runs then each go through many keys. On Kubernetes, a run of a pipeline with `tasks` gets a cluster of `workers` dista workers (`orca.kube`); locally it forks `ORCA_WORKERS`.

The scheduler runs at most `max_runs` of its runs at once, and starts a
failed scheduled or backfilled run again up to `run_retries` times, with
backoff: `retry_seconds` (0 for the scheduler's `ORCA_RETRY_SECONDS`)
doubling with each start, up to `max_retry_seconds` if it's set
(`orca.scheduler`). `run_retries=72, retry_seconds=3600,
max_retry_seconds=3600` tries hourly for three days.

Implements: `Movable`

#### `K`

```mojo
comptime K = partitions.K
```

The partition key.

#### `check`

```mojo
def check()
```

Asserts at compile time that the pipeline is well formed.

#### `registry_hash`

```mojo
def registry_hash() -> UInt64
```

A hash over the pipeline's name, its steps' names, kinds, edges, functions and datasets, and its dista tasks: two binaries with the same hash run the same DAG.

#### `describe`

```mojo
def describe() -> String
```

The DAG as JSON: the pipeline's name, its registry hash, its partitions, each step's name, kind, upstream, function, types and datasets in run order, and its dista tasks.

#### `timeouts`

```mojo
def timeouts() -> String
```

`step=seconds` for each step with `timeout_seconds`, comma-separated, for the run's watchdog (`orca.heartbeat`).

#### `sensed_datasets`

```mojo
def sensed_datasets() -> List[String]
```

The datasets its steps read and none of them writes, once each, in step order, and a keyed pipeline's keys dataset: what a sensor watches.

#### `scheduled`

```mojo
def scheduled() -> Bool
```

#### `load_schedule`

```mojo
def load_schedule() -> Schedule
```

The schedule, parsed, with its zone loaded; raises for a pipeline without one, or one whose expression or zone is bad.

#### `fills_for`

```mojo
def fills_for(fire: Int64, s: Schedule) -> Self.K
```

The partition a fire at `fire` fills.

#### `run`

```mojo
def run(keys: List[Self.K], run: Int = Int(0)) -> RunResult
```

Runs every step over `keys`, in order, recorded in a `MemoryStore` of its own, with a dista runtime of its own if the pipeline has tasks.

#### `execute`

```mojo
def execute[S: Store](mut store: S, var prior: RunResult, keys: List[Self.K], run: Int, runtime: Int = Int(0)) -> RunResult
```

Runs the steps over `keys` that haven't succeeded in `prior`, in order, reporting each attempt to `store`.

`prior` holds the steps a run had already finished when it's resumed,
with their outputs; they aren't run again. A step that raises is
retried up to its `retries`, then fails its partition. An error from
the store itself is raised: the run stops where it is, and can be
resumed from there. `runtime` is the address of the run's
`tasks.RT`, which steps reach through `Ctx.runtime`.