morca

morca · API

orca.store

Where the executor records what it does.

Pipeline.execute reports every attempt it starts and finishes, and every step it skips, to a Store; a step's reads and commits reach the same store through the Recorder its Ctx holds. MemoryStore keeps them in the process, for a run that lives as long as it does; orca.state.PgStore writes them to Postgres, so a run can be resumed by another process.

RunStore is a Store that also keeps runs: the rest of what a pipeline binary (orca.cli) asks of persistence, from registering the pipeline to ending its run, so the binary depends on this trait and nothing else. The scheduler and the UI query orca.state.State's SQL directly.

Structs

StepRun

struct StepRun

One step for one partition (or for the run, for a gather step).

Implements Copyable, Movable

Fields
step: String
partition: String: Empty for a gather step.
status: String
output: List[UInt8]: The encoded output, when the step succeeded.
error: String
attempt: Int: The number of the attempt that settled it; 0 for a skipped step.
until: Int64: When a deferred step may run again, on orca.civil.now's clock.

__init__

def __init__(out self, step: String, partition: String, status: String, var output: List[UInt8], error: String, attempt: Int = Int(0), until: Int64 = Int64(0))

RunResult

struct RunResult

What a run did: every step and partition, in the order they ran.

Implements Movable

Fields
steps: List[StepRun]

__init__

def __init__(out self)

succeeded

def succeeded(self) -> Bool

deferred_until

def deferred_until(self) -> Int64

When the run may go on, if a step deferred and none failed: the latest time its deferred steps asked for. 0 otherwise.

outcome

def outcome(self) -> String

succeeded, deferred or failed.

find

def find(self, step: StringSpan, partition: StringSpan) -> Int

The index of step for partition, or -1.

status

def status(self, step: StringSpan, partition: StringSpan) -> String

output

def output[T: Movable](self, step: StringSpan, partition: StringSpan = "") -> T

The decoded output of step for partition.

Backoff

struct Backoff

The waits between retries of one write to a store.

again(e) is called with what the write raised: it raises again if the error is final (FINAL) or the time (db_retry_seconds) is up, else it waits, a quarter of a second doubling to 4 s; the store then reconnects and tries again.

Implements Movable

Fields
what: String: The store, for the message: Postgres, SQLite.
started: Int
wait: Float64
budget: Float64

__init__

def __init__(out self, what: String)

again

def again(mut self, e: Error)

RunInfo

struct RunInfo

Implements Copyable, Movable

Fields
id: Int
pipeline: String
registry_hash: String
partitions: String
status: String
owner: String
error: String: Why the run ended without its steps saying so, or empty.

Usage

struct Usage

What an attempt's processes used (orca.usage): the peak resident bytes of its process tree, its CPU seconds, and the most cores it kept busy over a sample's interval. Each is -1 where it isn't known: a run on EC2 or Kubernetes, one no scheduler or worker started, or on macOS the peak cores, which only samples give.

Implements Copyable, ImplicitlyCopyable, Movable

Fields
peak_rss_bytes: Int
cpu_seconds: Float64
peak_cpus: Float64

unknown

def unknown() -> Self

known

def known(self) -> Bool

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.
usage: Usage: What the latest attempt used, as far as it's known.

TableSnapshot

struct TableSnapshot

A table, by its location, at a snapshot.

Implements Copyable, Movable

Fields
table: String
snapshot_id: Int64: -1 for a table that had no snapshot yet.
dataset: String: The table's dataset name, namespace.table, or empty for a table outside the warehouse.

Recorder

struct Recorder

Records an attempt's reads and commits in the store it came from, as they happen, so a process killed right after a commit still leaves it on record.

It holds the store by address, as Ctx holds the run's dista runtime: Ctx isn't generic over the store, and the store outlives every Ctx Pipeline.execute makes from it.

Implements Movable

__init__

def __init__(out self)

A recorder that records nothing.

def __init__[S: Store](out self, mut store: S)

A recorder that records in store.

fence

def fence(mut self)

Raises unless this process may still commit for its run: what orca.tables checks right before each commit (Store.fence).

read

def read(mut self, t: TableSnapshot)

commit

def commit(mut self, t: TableSnapshot)

MemRecord

struct MemRecord

A read or commit an attempt recorded in a MemoryStore.

Implements Copyable, Movable

Fields
attempt: Int: The attempt's index in the store's attempts.
commit: Bool
table: TableSnapshot

MemoryStore

struct MemoryStore

Keeps runs in the process's memory: what <pipeline> run uses without ORCA_DATABASE_URL, and Pipeline.run.

The same rows PgStore keeps, and the same rules: a run in it can be interrupted, claimed by another owner and resumed, so long as the process lives. Its runs are numbered 0, -1, -2, ...: never positive, so Ctx.made recognises none of their commits, since another process's in-memory run would have the same number. Nothing beats the heartbeat: no other process can see the run.

Implements Movable, RunStore, Store

Fields
records: List[MemRecord]: Every read and commit recorded, in order.
run: Int
owner: String
registry_hash: String

__init__

def __init__(out self)

request_cancel

def request_cancel(mut self, run: Int)

Asks run to cancel, as orca ui does in Postgres.

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)

start_heartbeat

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

stop_heartbeat

def stop_heartbeat(mut self)

finish_run

def finish_run(mut self, status: String)

defer_run

def defer_run(mut self, until: Int64)

cancel_requested

def cancel_requested(mut self) -> Bool

start_attempt

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

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)

pinned_reads

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

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)

Traits

Store

start_attempt

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

Records that step for partition is starting, with its encoded input, and returns the attempt's number (1 for the first). ord is the step's place in the run order.

finish_attempt

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

Records how attempt attempt ended: succeeded with its encoded output, or failed with its error.

skip

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

Records that step for partition didn't run, and why.

pinned_reads

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

The table snapshots earlier attempts of step for partition read, each table at its first attempt's snapshot.

run_token

def run_token(mut self) -> String

The run's random token, made with the run and the same for all its attempts, which goes into the summaries of the snapshots it commits (Ctx.token); empty for a run without one: an in-memory run, or one created before tokens.

record

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

Records that the attempt last started read t (or, if commit, committed it). What a Recorder from this store calls.

fence

def fence(mut self)

Raises unless this process still owns the run, with a heartbeat younger than half of ORCA_STALE_SECONDS, so no other process can have claimed it nor claim it during a commit: what a commit to Iceberg checks right before it's made (orca.ctx). Its message starts orca: not committing.

recorder

def recorder(mut self) -> Recorder

What records the reads and commits of the attempt just started.

set_log

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

Records where attempt attempt writes its log.

RunStore

A Store that keeps runs too: everything a pipeline binary records.

One process is a run's writer, its owner (process_owner): it creates or claims the run, owns it, which points the Store methods at it, keeps its heartbeat fresh, and ends it with finish_run. A store that other processes share stops an owner that lost its run, or whose run was asked to cancel, at its next attempt.

Implements Store

migrate

def migrate(mut self)

Creates the store's schema, or brings it up to date.

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 names, in image is; recorded the first time.

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.

claim

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

Makes owner the run's writer if no live process holds it: the run is queued, failed, or running with a heartbeat older than stale_seconds or cleared. Raises otherwise, saying why.

ran_version

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

Records that run is running version's code.

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: the process running it stopped.

prior

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

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

run_info

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

The run; raises if there's no such run.

runs

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

The latest limit runs, newest first; with before (unless 0), those older than run before (with ids under it, but in memory): the next page, keyed by the last run of the one before. With status, a status or several comma-separated, only runs with one of them.

steps

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

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

own

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

Records this process's attempts, as owner with code registry_hash, against run, which it has created or claimed.

start_heartbeat

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

Keeps the owned run's heartbeat fresh, every interval seconds, from a thread, until stop_heartbeat. The thread is also the run's watchdog (orca.heartbeat): timeouts is step=seconds for each step with timeout_seconds, comma-separated (Pipeline.timeouts).

stop_heartbeat

def stop_heartbeat(mut self)

Stops the heartbeat thread, if it's going, and waits for it.

finish_run

def finish_run(mut self, status: String)

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

defer_run

def defer_run(mut self, until: Int64)

Ends the owned run deferred, to go on at until (on orca.civil.now's clock): a step raised defer and none failed. That start isn't counted against its retries.

cancel_requested

def cancel_requested(mut self) -> Bool

Whether the owned run was asked to cancel.

Functions

db_retry_seconds

def db_retry_seconds() -> Float64

ORCA_DB_RETRY_SECONDS: how long a run's process keeps retrying a write to its store that failed, by default the stale limit (ORCA_STALE_SECONDS, 30), past which another process may take the run over anyway.

process_owner

def process_owner() -> String

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

status_list

def status_list(status: String) -> List[String]

runs' status, a status or several comma-separated, as a list, each once; empty for any.

Aliases

comptime RUNNING = "running"
comptime CANCELLED = "cancelled"
comptime SUCCEEDED = "succeeded"
comptime FAILED = "failed"
comptime SKIPPED = "skipped"
comptime DEFERRED = "deferred"

A step that raised defer: not done, and not failed.

comptime INTERRUPTED = "interrupted: the process running it stopped"

The error of an attempt whose process died without saying why.

comptime FINAL = "\1F"

Starts the message of an error a retry can't help: the run was lost to another process, or cancelled. Backoff raises it at once, without it.