# orca.cli

> A module of morca's API.

`orca_main`: the command line every pipeline binary gets.

```
taxi describe                              the DAG as JSON
taxi run --partitions 2024-01..2024-03     starts a run and runs it
taxi run ... --allow-new-code              ... with code the scheduler wasn't given
taxi resume <run> [--allow-new-code]       continues a run at what's unfinished
taxi status [<run>]                        the latest runs, or one run's steps
taxi status --before <run>                 the 20 runs older than that one
taxi status --status failed,queued         the latest runs with one of those statuses
taxi sense                                 the current snapshot of each dataset it reads
labels keys                                a keyed pipeline's keys snapshot, then its keys
```

For a keyed pipeline (`orca.keyed`), `--partitions` takes names: a list,
`@all` or `@file:<path>`. A name its keys dataset doesn't have now is
refused, as a malformed key is, with exit code 2.

With `ORCA_DATABASE_URL` set, runs are recorded in Postgres and can be
resumed, or, for `sqlite:<path>` (`sqlite:///abs/path` for an absolute
one), in that SQLite file (`orca.sqlite_store`); without it, `run` records its run in memory (`MemoryStore`, run 0)
on the same path, and `resume` and `status` aren't available. With
`ORCA_STATE_URL` (and `ORCA_STATE_TOKEN`) instead, as `orca worker` starts
a run, `resume`, `status <run>` and `keep-log` record over `orca ui`'s API
(`orca.http_store`), and `run` is refused. `run` records the pipeline the first time; once `orca register`
has given it a binary, `run` leaves the registration as it is, since what
the scheduler queues and counts as done is that code's. A binary whose code
isn't the registered one's is refused (exit code 2) unless `run` is given
`--allow-new-code`; its run then counts as that code's, not the
pipeline's. `ORCA_STALE_SECONDS` (default 30) is how old a run's heartbeat must
be before another process may resume it.

A pipeline with dista tasks gets a runtime per run, started before anything
else so its forked processes inherit no connection or thread; run with
`DISTA_ROLE` set, the binary is a dista head or worker instead (see
`orca.tasks`). With `ORCA_LOGS` set, a directory or a location in object
storage (`orca.logs`), attempts log there, the run's dista timeline is
written beside them, and `status <run>` shows each step's latest log.

`sense` prints a line per dataset the pipeline reads and doesn't write,
`<dataset> <snapshot> <orca|other> <partitions|->`: its table's current
snapshot in `ORCA_WAREHOUSE` (-1 for none), whether an orca step made it,
and the partition keys its summary says it holds. The scheduler polls it for
a sensor pipeline (`orca.sensors`).

`run` and `resume` print one line per step and partition, then the run's
status. A run whose step raised `defer` (and none failed) ends `deferred`,
and the scheduler starts it again when the step asked. Exit codes: 0 when
the run succeeded, 1 when a step didn't, the run deferred or was cancelled,
2 for a
usage error, 3 for any other error (the database, a run that can't be
resumed).

## Functions

### `seed_random`

```mojo
def seed_random()
```

Seeds `std.random` from the clock and the process ID.

Unseeded, every process starts from the same state, so pipeline
processes committing to one table at once draw the same Iceberg snapshot
IDs and file UUIDs (iceberg takes both from `std.random` and doesn't
seed it), and overwrite each other's manifest lists.

### `orca_main`

```mojo
def orca_main[P: AnyStruct[Pipeline[P.name, P.Ts, tasks=P.tasks, partitions=P.partitions, schedule=P.schedule, fills=P.fills, catchup=P.catchup, max_runs=P.max_runs, run_retries=P.run_retries, retry_seconds=P.retry_seconds, max_retry_seconds=P.max_retry_seconds, sensor=P.sensor, keys_per_run=P.keys_per_run, workers=P.workers]]]()
```

Parses the command line and runs the command on `P`.

### `report`

```mojo
def report(res: RunResult)
```

One line per step and partition, then the run's status.

## Aliases

### `USAGE`

```mojo
comptime USAGE = "usage:\n  <pipeline> describe\n  <pipeline> run --partitions KEY[..KEY][,...] [--allow-new-code]   (YYYY-MM, or YYYY-MM-DD daily)\n  <pipeline> run --partitions NAME[,...]|@all|@file:<path> [--allow-new-code]   (keyed)\n  <pipeline> resume <run> [--allow-new-code]\n  <pipeline> status [<run> | --before <run>] [--status <s>[,<s>...]]\n  <pipeline> sense\n  <pipeline> keys\n  <pipeline> fires [<count>] [--after YYYY-MM-DDTHH:MM[:SS]Z]\n  <pipeline> sample --out <location>|- [--pid PID] [--seconds S] ..."
```