# orca.http_store > A module of morca's API. `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 `/api/worker/store/` with `Authorization: Bearer `, 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 ` 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` ```mojo struct Body ``` A JSON object, written a field at a time: what a call sends. Implements: `Movable` Fields: - `out: String` #### `__init__` ```mojo def __init__(out self) ``` #### `text` ```mojo def text(var self, k: StringSpan, v: StringSpan) -> Self ``` #### `int` ```mojo def int(var self, k: StringSpan, v: Int64) -> Self ``` #### `float` ```mojo def float(var self, k: StringSpan, v: Float64) -> Self ``` #### `bool` ```mojo def bool(var self, k: StringSpan, v: Bool) -> Self ``` #### `bytes` ```mojo def bytes(var self, k: StringSpan, v: Span[UInt8]) -> Self ``` #### `raw` ```mojo def raw(var self, k: StringSpan, json: StringSpan) -> Self ``` `json`, already JSON, as the value. #### `finish` ```mojo def finish(deinit self) -> String ``` ### `StateApi` ```mojo struct StateApi ``` The UI's store calls, as a worker token. Implements: `Copyable`, `Movable` Fields: - `url: String` - `token: String` - `http: HttpClient` #### `__init__` ```mojo def __init__(out self, url: String, token: String) ``` #### `post_once` ```mojo 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` ```mojo def request_once(self, path: String, body: String) -> Value ``` `POST path` once, as `post_once`. #### `post` ```mojo def post(self, call: String, body: String) -> Value ``` `call`, retried while it fails but isn't refused for good. ### `HttpBeat` ```mojo 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__` ```mojo def __init__(out self, var api: StateApi, run: Int, owner: String, timeouts: String, parent: Int) ``` #### `beat` ```mojo def beat(mut self) ``` #### `reset` ```mojo def reset(mut self) ``` ### `HttpStore` ```mojo 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__` ```mojo def __init__(out self, url: String, token: String) ``` #### `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] ``` #### `run_token` ```mojo def run_token(mut self) -> String ``` #### `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) ``` ## Functions ### `hex_of` ```mojo def hex_of(b: Span[UInt8]) -> String ``` `b` in lower-case hex, as encoded inputs and outputs travel. ### `bytes_of_hex` ```mojo def bytes_of_hex(s: String) -> List[UInt8] ``` The bytes `hex_of` gave `s`; raises for anything else. ### `run_info_json` ```mojo def run_info_json(r: RunInfo) -> String ``` A `RunInfo` as the UI answers it. ### `run_info_of` ```mojo def run_info_of(v: Value) -> RunInfo ``` ### `steps_json` ```mojo def steps_json(steps: List[StepInfo]) -> String ``` ### `steps_of` ```mojo def steps_of(v: Value) -> List[StepInfo] ``` ### `prior_json` ```mojo def prior_json(res: RunResult) -> String ``` A run's succeeded steps and their outputs, for resuming it. ### `prior_of` ```mojo def prior_of(v: Value) -> RunResult ``` ### `pins_json` ```mojo def pins_json(pins: Dict[String, Int64]) -> String ``` ### `pins_of` ```mojo def pins_of(v: Value) -> Dict[String, Int64] ``` ### `refusal` ```mojo def refusal(text: String) -> String ``` The `error` of an answer's JSON, or empty. ### `check_state_url` ```mojo 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 ### `STORE_PATH` ```mojo comptime STORE_PATH = "/api/worker/store/" ``` Where a store call goes on the UI: this and the call's name.