# orca.state

> A module of Orca's API.

orca's state in Postgres: pipelines, runs, steps and attempts.

Everything lives in the `orca` schema, which `State.migrate` creates and
brings up to date; migrations are SQL in `_migrations`, applied in order
under a transaction-scoped advisory lock, so two processes starting at once
don't both apply one.

**One writer per run.** The process running a run is its `owner` and keeps
`runs.heartbeat` fresh from a thread (`orca.heartbeat`). `State.claim` takes
a run only if its heartbeat is older than the stale limit, or cleared, which
`finish_run` does when a run ends. Every attempt checks in the same
transaction that this process still owns the run, so a process that lost its
run to a resume stops at its next step instead of racing it. The same check
stops a run that was asked to cancel (`runs.cancel_requested`, which `orca
ui` sets) before its next attempt; an attempt already going finishes and is
recorded.

**Each change is one transaction.** Starting an attempt marks its step
running and inserts the attempt; finishing it records the output or the
error on both. A process killed in between leaves an attempt `running`,
which `interrupt` marks failed when the run is resumed.

## Structs

### `RunInfo`

```mojo
struct RunInfo
```

Implements: `Copyable`, `Movable`

Fields:

- `id: Int`
- `pipeline: String`
- `registry_hash: String`
- `partitions: String`
- `status: String`
- `owner: String`

### `StepInfo`

```mojo
struct StepInfo
```

Implements: `Copyable`, `Movable`

Fields:

- `step: String`
- `partition: String`
- `status: String`
- `attempts: Int`
- `error: String`
- `log: String`: The latest attempt's log file, or empty.

### `State`

```mojo
struct State
```

A connection to orca's database.

Implements: `Movable`

Fields:

- `conn: Connection`

#### `__init__`

```mojo
def __init__(out self, url: String)
```

#### `migrate`

```mojo
def migrate(mut self)
```

Creates the schema, or applies the migrations it hasn't had.

#### `register`

```mojo
def register(mut self, pipeline: String, registry_hash: String, describe: String)
```

Records a pipeline, or updates it to this binary's DAG.

#### `version`

```mojo
def version(mut self, pipeline: String, registry_hash: String, describe: String, image: String) -> Int
```

The version of `pipeline` that code `registry_hash`, from the source `describe` (a pipeline binary's) names, in `image` is; recorded the first time it's seen.

#### `ran_version`

```mojo
def ran_version(mut self, run: Int, version: Int)
```

Records that `run` is running `version`'s code.

#### `create_run`

```mojo
def create_run(mut self, pipeline: String, registry_hash: String, partitions: String, trigger: String, owner: String) -> Int
```

A new run, already claimed by `owner`.

#### `run_info`

```mojo
def run_info(mut self, run: Int) -> RunInfo
```

#### `claim`

```mojo
def claim(mut self, run: Int, owner: String, stale_seconds: Float64)
```

Makes `owner` the run's writer, if no live process holds it.

A run can be claimed while it's `queued`, unfinished (`running` with
a stale or cleared heartbeat; the scheduler starts a run with none)
or `failed`. Raises otherwise, saying why.

#### `interrupt`

```mojo
def interrupt(mut self, run: Int) -> Int
```

Marks attempts left `running` by a process that died as failed, and returns how many there were.

#### `prior`

```mojo
def prior(mut self, run: Int) -> RunResult
```

The run's succeeded steps, with their outputs, for resuming it.

#### `finish_run`

```mojo
def finish_run(mut self, run: Int, owner: String, status: String)
```

Ends the run and clears its heartbeat, so it can be resumed at once if it failed.

#### `cancel_requested`

```mojo
def cancel_requested(mut self, run: Int) -> Bool
```

#### `runs`

```mojo
def runs(mut self, limit: Int = Int(20)) -> List[RunInfo]
```

The latest runs, newest first.

#### `steps`

```mojo
def steps(mut self, run: Int) -> List[StepInfo]
```

A run's steps in run order, with their attempt counts.

### `PgStore`

```mojo
struct PgStore
```

Records a run's attempts in Postgres, as its owner.

Implements: `Movable`, `Store`

Fields:

- `db: State`
- `url: String`
- `run: Int`
- `owner: String`
- `registry_hash: String`

#### `__init__`

```mojo
def __init__(out self, var db: State, url: String, run: Int, owner: String, registry_hash: String)
```

#### `start_attempt`

```mojo
def start_attempt(mut self, step: String, partition: String, ord: Int, input: Span[UInt8]) -> Int
```

#### `pinned_reads`

```mojo
def pinned_reads(mut self, step: String, partition: String) -> Dict[String, Int64]
```

#### `recorder`

```mojo
def recorder(mut self) -> Recorder
```

#### `set_log`

```mojo
def set_log(mut self, step: String, partition: String, attempt: Int, path: String)
```

#### `finish_attempt`

```mojo
def finish_attempt(mut self, step: String, partition: String, attempt: Int, status: String, output: Span[UInt8], error: String)
```

#### `skip`

```mojo
def skip(mut self, step: String, partition: String, ord: Int, reason: String)
```

## Functions

### `process_owner`

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

`host:pid`, naming this process as a run's owner.

## Aliases

### `RUNNING`

```mojo
comptime RUNNING = "running"
```

### `CANCELLED`

```mojo
comptime CANCELLED = "cancelled"
```