# orca.state > A module of morca'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 ### `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. #### `scheduled_hash` ```mojo def scheduled_hash(mut self, pipeline: String) -> String ``` The registry hash `orca register` gave `pipeline`, or empty if it wasn't registered with a binary. #### `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, why: String = "") -> Int ``` Marks attempts left `running` by a process that died as failed, and returns how many there were. `why` is their error and their steps', such as a crash's (`orca.crash`); empty, it's `INTERRUPTED`. #### `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. #### `defer_run` ```mojo def defer_run(mut self, run: Int, owner: String, until: Int64) ``` Ends the run `deferred`, to be started again at `until` (on the scheduler's clock) by the scheduler, whatever started it; that start isn't counted against its retries. #### `cancel_requested` ```mojo def cancel_requested(mut self, run: Int) -> Bool ``` #### `runs` ```mojo def runs(mut self, limit: Int = Int(20), before: Int = Int(0), status: String = "") -> List[RunInfo] ``` The latest runs, newest first; those under `before` if it isn't 0, and with `status` (one or several, comma-separated) one of those. #### `steps` ```mojo def steps(mut self, run: Int) -> List[StepInfo] ``` A run's steps in run order, with their attempt counts. ### `DbRetry` ```mojo struct DbRetry ``` Retries of one call to Postgres from a run's process: `Backoff`, then a new connection, so a restart or failover of the server doesn't stop the run. Implements: `Movable` Fields: - `backoff: Backoff` #### `__init__` ```mojo def __init__(out self) ``` #### `again` ```mojo def again(mut self, e: Error, mut db: State, url: String) ``` ### `PgBeat` ```mojo struct PgBeat ``` Refreshes a run's heartbeat in Postgres, on a connection of its own. Implements: `Beater`, `Movable` Fields: - `url: String` - `run: Int` - `owner: String` - `timeouts: String`: `step=seconds` for each step with a timeout, comma-separated. #### `__init__` ```mojo def __init__(out self, url: String, run: Int, owner: String, timeouts: String = "") ``` #### `beat` ```mojo def beat(mut self) ``` #### `reset` ```mojo def reset(mut self) ``` ### `PgStore` ```mojo struct PgStore ``` orca's runs in Postgres, `State` behind the `RunStore` trait. The calls for the owned run (attempts, skips, reads and commits, its end) are retried for `ORCA_DB_RETRY_SECONDS` on a new connection (`DbRetry`), except when the run was lost or cancelled, so a blip of the server doesn't throw away a step that finished. The others are made once, before the run starts or after it ends. Implements: `Movable`, `RunStore`, `Store` Fields: - `db: State` - `url: String` - `run: Int` - `owner: String` - `registry_hash: String` #### `__init__` ```mojo def __init__(out self, url: String) ``` Connects to `url`; `own` a run before recording attempts. ```mojo def __init__(out self, var db: State, url: String, run: Int, owner: String, registry_hash: String) ``` Over `db`, already owning `run`. #### `migrate` ```mojo def migrate(mut self) ``` #### `register` ```mojo def register(mut self, pipeline: String, registry_hash: String, describe: String) ``` #### `scheduled_hash` ```mojo def scheduled_hash(mut self, pipeline: String) -> String ``` #### `version` ```mojo def version(mut self, pipeline: String, registry_hash: String, describe: String, image: String) -> Int ``` #### `create_run` ```mojo def create_run(mut self, pipeline: String, registry_hash: String, partitions: String, trigger: String, owner: String) -> Int ``` #### `claim` ```mojo def claim(mut self, run: Int, owner: String, stale_seconds: Float64) ``` #### `ran_version` ```mojo def ran_version(mut self, run: Int, version: Int) ``` #### `interrupt` ```mojo def interrupt(mut self, run: Int, why: String = "") -> Int ``` #### `prior` ```mojo def prior(mut self, run: Int) -> RunResult ``` #### `run_info` ```mojo def run_info(mut self, run: Int) -> RunInfo ``` #### `runs` ```mojo def runs(mut self, limit: Int, before: Int = Int(0), status: String = "") -> List[RunInfo] ``` #### `steps` ```mojo def steps(mut self, run: Int) -> List[StepInfo] ``` #### `own` ```mojo def own(mut self, run: Int, owner: String, registry_hash: String) ``` #### `attempt_id` ```mojo def attempt_id(self) -> Int64 ``` The `attempts` row of the attempt last started: what a worker's run is told, to name it in its later calls (`orca.workers`). #### `at_attempt` ```mojo def at_attempt(mut self, id: Int64) ``` Points `record` and `set_log` at the `attempts` row `id`. #### `start_attempt_once` ```mojo def start_attempt_once(mut self, step: String, partition: String, ord: Int, input: Span[UInt8]) -> Int ``` `start_attempt` without its retries, its errors as they are (`FINAL` for a lost or cancelled run): for `orca ui`, whose worker retries the call itself (`orca.http_store`). #### `finish_attempt_once` ```mojo def finish_attempt_once(mut self, step: String, partition: String, attempt: Int, status: String, output: Span[UInt8], error: String) ``` `finish_attempt` without its retries, as `start_attempt_once`. #### `skip_once` ```mojo def skip_once(mut self, step: String, partition: String, ord: Int, reason: String) ``` `skip` without its retries, as `start_attempt_once`. #### `start_heartbeat` ```mojo def start_heartbeat(mut self, interval: Float64, timeouts: String = "") ``` #### `stop_heartbeat` ```mojo def stop_heartbeat(mut self) ``` #### `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] ``` #### `run_token` ```mojo def run_token(mut self) -> String ``` #### `record` ```mojo def record(mut self, commit: Bool, t: TableSnapshot) ``` #### `fence` ```mojo def fence(mut self) ``` #### `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) ``` #### `finish_run` ```mojo def finish_run(mut self, status: String) ``` `State.finish_run` for this run, retried as the attempts are. #### `defer_run` ```mojo def defer_run(mut self, until: Int64) ``` `State.defer_run` for this run, retried as the attempts are. #### `cancel_requested` ```mojo def cancel_requested(mut self) -> Bool ``` ## Functions ### `live_older` ```mojo def live_older(mut tx: Transaction, floor: Int) -> List[Tuple[Int, String, Int]] ``` The scheduler instances (docs/overlap.md) still going, seen within the stale limit and not ended, that know fewer than `floor` migrations: each one's id, host and how many it knows. None where `orca.schedulers` doesn't exist yet. ### `beat_run` ```mojo def beat_run(mut conn: Connection, run: Int, owner: String, timeouts: String) -> String ``` One beat of `run`'s heartbeat as `owner`, and the watchdog's look at it (`orca.heartbeat`): why the run's process must stop now, a cancel whose time has come or an attempt past its step's timeout, which is then recorded, or empty. `PgBeat`'s, and `orca ui`'s for a worker's run (`orca.workers`), whose process then stops itself.