# orca.store > A module of morca's API. 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` ```mojo 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__` ```mojo 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` ```mojo struct RunResult ``` What a run did: every step and partition, in the order they ran. Implements: `Movable` Fields: - `steps: List[StepRun]` #### `__init__` ```mojo def __init__(out self) ``` #### `succeeded` ```mojo def succeeded(self) -> Bool ``` #### `deferred_until` ```mojo 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` ```mojo def outcome(self) -> String ``` `succeeded`, `deferred` or `failed`. #### `find` ```mojo def find(self, step: StringSpan, partition: StringSpan) -> Int ``` The index of `step` for `partition`, or -1. #### `status` ```mojo def status(self, step: StringSpan, partition: StringSpan) -> String ``` #### `output` ```mojo def output[T: Movable](self, step: StringSpan, partition: StringSpan = "") -> T ``` The decoded output of `step` for `partition`. ### `Backoff` ```mojo 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__` ```mojo def __init__(out self, what: String) ``` #### `again` ```mojo def again(mut self, e: Error) ``` ### `RunInfo` ```mojo 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` ```mojo 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` ```mojo def unknown() -> Self ``` #### `known` ```mojo def known(self) -> Bool ``` ### `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. - `usage: Usage`: What the latest attempt used, as far as it's known. ### `TableSnapshot` ```mojo 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` ```mojo 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__` ```mojo def __init__(out self) ``` A recorder that records nothing. ```mojo def __init__[S: Store](out self, mut store: S) ``` A recorder that records in `store`. #### `fence` ```mojo 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` ```mojo def read(mut self, t: TableSnapshot) ``` #### `commit` ```mojo def commit(mut self, t: TableSnapshot) ``` ### `MemRecord` ```mojo 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` ```mojo struct MemoryStore ``` Keeps runs in the process's memory: what ` 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__` ```mojo def __init__(out self) ``` #### `request_cancel` ```mojo def request_cancel(mut self, run: Int) ``` Asks `run` to cancel, as `orca ui` does in Postgres. #### `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) ``` #### `start_heartbeat` ```mojo def start_heartbeat(mut self, interval: Float64, timeouts: String = "") ``` #### `stop_heartbeat` ```mojo def stop_heartbeat(mut self) ``` #### `finish_run` ```mojo def finish_run(mut self, status: String) ``` #### `defer_run` ```mojo def defer_run(mut self, until: Int64) ``` #### `cancel_requested` ```mojo def cancel_requested(mut self) -> Bool ``` #### `start_attempt` ```mojo def start_attempt(mut self, step: String, partition: String, ord: Int, input: Span[UInt8]) -> Int ``` #### `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) ``` #### `pinned_reads` ```mojo def pinned_reads(mut self, step: String, partition: String) -> Dict[String, Int64] ``` #### `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) ``` ## Traits ### `Store` ```mojo undefined ``` #### `start_attempt` ```mojo 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` ```mojo 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` ```mojo def skip(mut self, step: String, partition: String, ord: Int, reason: String) ``` Records that `step` for `partition` didn't run, and why. #### `pinned_reads` ```mojo 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` ```mojo 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` ```mojo 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` ```mojo 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` ```mojo def recorder(mut self) -> Recorder ``` What records the reads and commits of the attempt just started. #### `set_log` ```mojo def set_log(mut self, step: String, partition: String, attempt: Int, path: String) ``` Records where attempt `attempt` writes its log. ### `RunStore` ```mojo undefined ``` 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, `own`s 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` ```mojo def migrate(mut self) ``` Creates the store's schema, or brings it up to date. #### `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` names, in `image` is; recorded the first time. #### `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`. #### `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: the run is `queued`, `failed`, or `running` with a heartbeat older than `stale_seconds` or cleared. Raises otherwise, saying why. #### `ran_version` ```mojo def ran_version(mut self, run: Int, version: Int) ``` Records that `run` is running `version`'s code. #### `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: the process running it stopped`. #### `prior` ```mojo def prior(mut self, run: Int) -> RunResult ``` The run's succeeded steps, with their outputs, for resuming it. #### `run_info` ```mojo def run_info(mut self, run: Int) -> RunInfo ``` The run; raises if there's no such run. #### `runs` ```mojo 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` ```mojo def steps(mut self, run: Int) -> List[StepInfo] ``` A run's steps in run order, with their attempt counts. #### `own` ```mojo 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` ```mojo 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` ```mojo def stop_heartbeat(mut self) ``` Stops the heartbeat thread, if it's going, and waits for it. #### `finish_run` ```mojo 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` ```mojo 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` ```mojo def cancel_requested(mut self) -> Bool ``` Whether the owned run was asked to cancel. ## Functions ### `db_retry_seconds` ```mojo 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` ```mojo def process_owner() -> String ``` `host:pid`, naming this process as a run's owner. ### `status_list` ```mojo def status_list(status: String) -> List[String] ``` `runs`' `status`, a status or several comma-separated, as a list, each once; empty for any. ## Aliases ### `RUNNING` ```mojo comptime RUNNING = "running" ``` ### `CANCELLED` ```mojo comptime CANCELLED = "cancelled" ``` ### `SUCCEEDED` ```mojo comptime SUCCEEDED = "succeeded" ``` ### `FAILED` ```mojo comptime FAILED = "failed" ``` ### `SKIPPED` ```mojo comptime SKIPPED = "skipped" ``` ### `DEFERRED` ```mojo comptime DEFERRED = "deferred" ``` A step that raised `defer`: not done, and not failed. ### `INTERRUPTED` ```mojo comptime INTERRUPTED = "interrupted: the process running it stopped" ``` The error of an attempt whose process died without saying why. ### `FINAL` ```mojo 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.