morca

morca · API

orca.http_store

HttpStore: a run's state over orca ui's API, for a run on a remote worker (orca worker, orca.worker), which holds no database credentials.

A pipeline binary given ORCA_STATE_URL (the UI, https://..., or http to this machine only) and ORCA_STATE_TOKEN (a worker token, orca.workers) in place of ORCA_DATABASE_URL records its run through this store: each RunStore call is a POST of JSON to <ORCA_STATE_URL>/api/worker/store/<call> with Authorization: Bearer <token>, which the UI answers by making the same call on orca.state.PgStore (orca.workers.store_call), so what is recorded, and every check made on the way (the run still this process's, not cancelled, its heartbeat fresh enough to commit), is the same code as for a run that writes to Postgres itself. Encoded inputs and outputs travel as hex.

Retries. A call that can't reach the UI, or that it answers with a 5xx, is retried as PgStore retries Postgres (orca.store.Backoff, for ORCA_DB_RETRY_SECONDS); one refused (401 or 403: the token was revoked, or the run is another worker's now; 409: the run was lost or cancelled) is final, and the run stops. A fence (Store.fence) that can't be made within half the stale limit refuses the commit, as over Postgres.

The heartbeat is a POST .../beat from a thread every beat, which the UI answers with why the run must stop now, if the watchdog says so (orca.state.beat_run); the process then exits, as over Postgres. With ORCA_WORKER_PID set, the beat also checks that the worker that started this process is still its parent: if it isn't (the worker was killed), the process exits rather than go on beating, so the server sees the run's heartbeat go stale and requeues it.

Only resume, status <run> and keep-log work over this store: a worker's run resumes the run it was given, so run, which creates one, and status without a run need the database.

Structs

Body

struct Body

A JSON object, written a field at a time: what a call sends.

Implements Movable

Fields
out: String

__init__

def __init__(out self)

text

def text(var self, k: StringSpan, v: StringSpan) -> Self

int

def int(var self, k: StringSpan, v: Int64) -> Self

float

def float(var self, k: StringSpan, v: Float64) -> Self

bool

def bool(var self, k: StringSpan, v: Bool) -> Self

bytes

def bytes(var self, k: StringSpan, v: Span[UInt8]) -> Self

raw

def raw(var self, k: StringSpan, json: StringSpan) -> Self

json, already JSON, as the value.

finish

def finish(deinit self) -> String

StateApi

struct StateApi

The UI's store calls, as a worker token.

Implements Copyable, Movable

Fields
url: String
token: String
http: HttpClient

__init__

def __init__(out self, url: String, token: String)

post_once

def post_once(self, call: String, body: String) -> Value

Store call call once: its answer, or raises, with FINAL first when the UI refused it for good (the token, or the run, is no longer this worker's: 401, 403, 409).

request_once

def request_once(self, path: String, body: String) -> Value

POST path once, as post_once.

post

def post(self, call: String, body: String) -> Value

call, retried while it fails but isn't refused for good.

HttpBeat

struct HttpBeat

A run's heartbeat over the UI, with a client of its own.

Implements Beater, Movable

Fields
api: StateApi
run: Int
owner: String
timeouts: String
parent: Int: ORCA_WORKER_PID: the worker that started this process, or 0.

__init__

def __init__(out self, var api: StateApi, run: Int, owner: String, timeouts: String, parent: Int)

beat

def beat(mut self)

reset

def reset(mut self)

HttpStore

struct HttpStore

A run's state over orca ui's API (ORCA_STATE_URL).

Implements Movable, RunStore, Store

Fields
api: StateApi
run: Int
owner: String
registry_hash: String

__init__

def __init__(out self, url: String, token: String)

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]

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)

Functions

hex_of

def hex_of(b: Span[UInt8]) -> String

b in lower-case hex, as encoded inputs and outputs travel.

bytes_of_hex

def bytes_of_hex(s: String) -> List[UInt8]

The bytes hex_of gave s; raises for anything else.

run_info_json

def run_info_json(r: RunInfo) -> String

A RunInfo as the UI answers it.

run_info_of

def run_info_of(v: Value) -> RunInfo

steps_json

def steps_json(steps: List[StepInfo]) -> String

steps_of

def steps_of(v: Value) -> List[StepInfo]

prior_json

def prior_json(res: RunResult) -> String

A run's succeeded steps and their outputs, for resuming it.

prior_of

def prior_of(v: Value) -> RunResult

pins_json

def pins_json(pins: Dict[String, Int64]) -> String

pins_of

def pins_of(v: Value) -> Dict[String, Int64]

refusal

def refusal(text: String) -> String

The error of an answer's JSON, or empty.

check_state_url

def check_state_url(url: String) -> String

url without a trailing / if it's https, or http to this machine, where nothing is on the wire; raises otherwise, since the token would cross a network in the clear.

Aliases

comptime STORE_PATH = "/api/worker/store/"

Where a store call goes on the UI: this and the call's name.