magmalake Orca

Orca · 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

RunInfo

struct RunInfo

Implements Copyable, Movable

Fields
id: Int
pipeline: String
registry_hash: String
partitions: String
status: String
owner: String

StepInfo

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

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.

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) -> Int

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

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.

cancel_requested

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

runs

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

The latest runs, newest first.

steps

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

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

PgStore

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__

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

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]

recorder

def recorder(mut self) -> Recorder

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)

Functions

process_owner

def process_owner() -> String

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

Aliases

comptime RUNNING = "running"
comptime CANCELLED = "cancelled"