morca

morca · API

orca.state

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

struct State

A connection to orca's database.

Implements Movable

Fields
conn: Connection

__init__

def __init__(out self, url: String)

migrate

def migrate(mut self)

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

register

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

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

scheduled_hash

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

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

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

Records that run is running version's code.

create_run

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

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

claim

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

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

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

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

finish_run

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

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

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

runs

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

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

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

DbRetry

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__

def __init__(out self)

again

def again(mut self, e: Error, mut db: State, url: String)

PgBeat

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__

def __init__(out self, url: String, run: Int, owner: String, timeouts: String = "")

beat

def beat(mut self)

reset

def reset(mut self)

PgStore

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__

def __init__(out self, url: String)

Connects to url; own a run before recording attempts.

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

Over db, already owning run.

migrate

def migrate(mut self)

register

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

scheduled_hash

def scheduled_hash(mut self, pipeline: String) -> String

version

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

create_run

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

claim

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

ran_version

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

interrupt

def interrupt(mut self, run: Int, why: String = "") -> Int

prior

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

run_info

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

runs

def runs(mut self, limit: Int, before: Int = Int(0), status: String = "") -> List[RunInfo]

steps

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

own

def own(mut self, run: Int, owner: String, registry_hash: String)

attempt_id

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

def at_attempt(mut self, id: Int64)

Points record and set_log at the attempts row id.

start_attempt_once

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

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

def skip_once(mut self, step: String, partition: String, ord: Int, reason: String)

skip without its retries, as start_attempt_once.

start_heartbeat

def start_heartbeat(mut self, interval: Float64, timeouts: String = "")

stop_heartbeat

def stop_heartbeat(mut self)

start_attempt

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

pinned_reads

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

run_token

def run_token(mut self) -> String

record

def record(mut self, commit: Bool, t: TableSnapshot)

fence

def fence(mut self)

set_log

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

finish_attempt

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

skip

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

finish_run

def finish_run(mut self, status: String)

State.finish_run for this run, retried as the attempts are.

defer_run

def defer_run(mut self, until: Int64)

State.defer_run for this run, retried as the attempts are.

cancel_requested

def cancel_requested(mut self) -> Bool

Functions

live_older

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

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.