orbit_runtime/store/
mod.rs

1//! Persistence.
2//!
3//! The `execution_events` table is the source of truth. `workflow_executions`
4//! is a projection kept in the same transaction for fast listing. Every batch
5//! of events is appended atomically together with the projection update.
6//! `execution_events` holds each execution's current run; runs that ended
7//! with continue-as-new are moved to `execution_run_events` (keyed by run) in
8//! the transaction that starts the next run. `orbit_schedules` holds the
9//! schedules (see `crate::schedule`).
10//!
11//! Two backends share the schema and the guarantees:
12//!
13//! * **SQLite** (default): a single file, WAL mode, `synchronous=FULL`.
14//! * **PostgreSQL**: pass a `postgres://` URL. All statements run on one
15//!   dedicated connection thread, so the synchronous engine never blocks the
16//!   async server runtime from inside it, and the connection is re-established
17//!   automatically if it drops.
18
19mod postgres;
20mod sqlite;
21
22use crate::model::{ExecutionState, HistoryEvent};
23use orbit_ir::WorkflowIr;
24use serde::Serialize;
25use std::path::Path;
26
27pub(crate) use self::postgres::{session_client, CLUSTER_SCHEMA};
28pub use self::postgres::{PgMode, PgStore};
29pub(crate) use self::postgres::{SharedAppend, TaskOffer, UnitSync, NOW_MS, SHARED_SCHEMA};
30pub use self::sqlite::SqliteStore;
31
32#[derive(Debug)]
33pub enum PutDefinition {
34    Created,
35    Unchanged,
36    Conflict { existing_hash: String },
37}
38
39#[derive(Debug, Clone, Serialize)]
40pub struct ExecutionSummary {
41    pub execution_id: String,
42    pub workflow: String,
43    pub version: String,
44    pub status: String,
45    pub created_at_ms: u64,
46    pub updated_at_ms: u64,
47}
48
49#[derive(Debug, Clone, Serialize)]
50pub struct WorkerRecord {
51    pub worker_id: String,
52    pub runtime: String,
53    pub sdk: Option<String>,
54    pub contracts: serde_json::Value,
55    pub last_seen_ms: u64,
56    pub registered_at_ms: u64,
57}
58
59/// A storage error from either backend.
60#[derive(Debug, Clone, PartialEq, Eq)]
61pub struct StoreError(pub String);
62
63impl std::fmt::Display for StoreError {
64    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
65        f.write_str(&self.0)
66    }
67}
68
69impl std::error::Error for StoreError {}
70
71impl From<rusqlite::Error> for StoreError {
72    fn from(e: rusqlite::Error) -> Self {
73        StoreError(format!("sqlite: {e}"))
74    }
75}
76
77impl From<::postgres::Error> for StoreError {
78    fn from(e: ::postgres::Error) -> Self {
79        StoreError(format!("postgres: {e}"))
80    }
81}
82
83pub type Result<T> = std::result::Result<T, StoreError>;
84
85/// Shared schema. `BIGINT` is rewritten to `INTEGER` for SQLite.
86pub(crate) const SCHEMA: &str = r#"
87CREATE TABLE IF NOT EXISTS schema_version (version BIGINT NOT NULL);
88CREATE TABLE IF NOT EXISTS workflow_definitions (
89    name TEXT NOT NULL,
90    version TEXT NOT NULL,
91    definition_hash TEXT NOT NULL,
92    ir TEXT NOT NULL,
93    created_at_ms BIGINT NOT NULL,
94    PRIMARY KEY (name, version)
95);
96CREATE TABLE IF NOT EXISTS workflow_executions (
97    execution_id TEXT PRIMARY KEY,
98    workflow TEXT NOT NULL,
99    version TEXT NOT NULL,
100    definition_hash TEXT NOT NULL,
101    status TEXT NOT NULL,
102    created_at_ms BIGINT NOT NULL,
103    updated_at_ms BIGINT NOT NULL,
104    last_seq BIGINT NOT NULL
105);
106CREATE INDEX IF NOT EXISTS idx_executions_status ON workflow_executions(status);
107CREATE TABLE IF NOT EXISTS execution_events (
108    execution_id TEXT NOT NULL,
109    seq BIGINT NOT NULL,
110    event_type TEXT NOT NULL,
111    step TEXT,
112    payload TEXT NOT NULL,
113    timestamp_ms BIGINT NOT NULL,
114    PRIMARY KEY (execution_id, seq)
115);
116CREATE TABLE IF NOT EXISTS worker_registry (
117    worker_id TEXT PRIMARY KEY,
118    runtime TEXT NOT NULL,
119    sdk TEXT,
120    contracts TEXT NOT NULL,
121    registered_at_ms BIGINT NOT NULL,
122    last_seen_ms BIGINT NOT NULL
123);
124CREATE TABLE IF NOT EXISTS execution_run_events (
125    execution_id TEXT NOT NULL,
126    run BIGINT NOT NULL,
127    seq BIGINT NOT NULL,
128    event_type TEXT NOT NULL,
129    step TEXT,
130    payload TEXT NOT NULL,
131    timestamp_ms BIGINT NOT NULL,
132    PRIMARY KEY (execution_id, run, seq)
133);
134CREATE TABLE IF NOT EXISTS orbit_schedules (
135    schedule_id TEXT PRIMARY KEY,
136    spec TEXT NOT NULL,
137    paused BIGINT NOT NULL,
138    next_fire_ms BIGINT,
139    due_ms BIGINT,
140    last_fire_ms BIGINT,
141    last_execution_id TEXT,
142    fired_count BIGINT NOT NULL,
143    skipped_count BIGINT NOT NULL,
144    last_error TEXT,
145    claimed_by TEXT,
146    claimed_until_ms BIGINT,
147    created_at_ms BIGINT NOT NULL,
148    updated_at_ms BIGINT NOT NULL
149);
150CREATE INDEX IF NOT EXISTS idx_orbit_schedules_due ON orbit_schedules(due_ms);
151"#;
152
153/// A schedule as stored in `orbit_schedules`. `spec` is the JSON of
154/// [`crate::schedule::ScheduleSpec`]; the other columns are the schedule's
155/// progress, written by whichever server fires it.
156#[derive(Debug, Clone, PartialEq)]
157pub struct ScheduleRow {
158    pub schedule_id: String,
159    pub spec: String,
160    pub paused: bool,
161    /// The next occurrence (nominal time); `None` once the schedule ended.
162    pub next_fire_ms: Option<u64>,
163    /// When that occurrence fires: `next_fire_ms` plus the jitter.
164    pub due_ms: Option<u64>,
165    pub last_fire_ms: Option<u64>,
166    pub last_execution_id: Option<String>,
167    pub fired: u64,
168    pub skipped: u64,
169    pub last_error: Option<String>,
170    pub created_at_ms: u64,
171    pub updated_at_ms: u64,
172}
173
174/// The outcome of firing one occurrence, written when the firing server
175/// releases its claim (see [`Store::finish_schedule_fire`]).
176#[derive(Debug, Clone, Default, PartialEq)]
177pub struct FireRecord {
178    pub next_fire_ms: Option<u64>,
179    pub due_ms: Option<u64>,
180    /// Set when an occurrence was handled (started or skipped).
181    pub last_fire_ms: Option<u64>,
182    /// Set when an execution was started.
183    pub last_execution_id: Option<String>,
184    pub fired: u64,
185    pub skipped: u64,
186    pub last_error: Option<String>,
187}
188
189/// Continue-as-new: the finished run's number and its last events. They are
190/// archived, together with the rest of that run's history, in the
191/// transaction that writes the next run's first events.
192pub struct Rollover<'a> {
193    pub run: u32,
194    pub events: &'a [HistoryEvent],
195}
196
197/// The storage backend used by the engine.
198pub enum Store {
199    Sqlite(SqliteStore),
200    Postgres(PgStore),
201}
202
203macro_rules! dispatch {
204    ($self:ident, $s:ident => $call:expr) => {
205        match $self {
206            Store::Sqlite($s) => $call,
207            Store::Postgres($s) => $call,
208        }
209    };
210}
211
212impl Store {
213    /// Open a backend from a location: a `postgres://` / `postgresql://` URL,
214    /// or a SQLite file path.
215    pub fn connect(location: &str) -> Result<Store> {
216        if location.starts_with("postgres://") || location.starts_with("postgresql://") {
217            Ok(Store::Postgres(PgStore::connect(location)?))
218        } else {
219            Store::open(Path::new(location))
220        }
221    }
222
223    /// Open (or create) a SQLite database file.
224    pub fn open(path: &Path) -> Result<Store> {
225        Ok(Store::Sqlite(SqliteStore::open(path)?))
226    }
227
228    pub fn in_memory() -> Result<Store> {
229        Ok(Store::Sqlite(SqliteStore::in_memory()?))
230    }
231
232    pub fn backend(&self) -> &'static str {
233        match self {
234            Store::Sqlite(_) => "sqlite",
235            Store::Postgres(_) => "postgres",
236        }
237    }
238
239    pub fn put_definition(&self, ir: &WorkflowIr, now: u64) -> Result<PutDefinition> {
240        dispatch!(self, s => s.put_definition(ir, now))
241    }
242
243    pub fn definitions(&self) -> Result<Vec<WorkflowIr>> {
244        dispatch!(self, s => s.definitions())
245    }
246
247    pub fn execution_exists(&self, id: &str) -> Result<bool> {
248        dispatch!(self, s => s.execution_exists(id))
249    }
250
251    /// Atomically append events and update the execution projection.
252    pub fn append(&mut self, state: &ExecutionState, events: &[HistoryEvent]) -> Result<()> {
253        dispatch!(self, s => s.append(state, events))
254    }
255
256    pub fn history(&self, id: &str) -> Result<Vec<HistoryEvent>> {
257        dispatch!(self, s => s.history(id))
258    }
259
260    /// Continue-as-new in one transaction: archive the finished run (its
261    /// stored history plus `rollover.events`) under its run number, then
262    /// write the next run's first `events` and point the projection at it.
263    pub fn append_rollover(
264        &mut self,
265        state: &ExecutionState,
266        events: &[HistoryEvent],
267        rollover: Rollover<'_>,
268    ) -> Result<()> {
269        dispatch!(self, s => s.append_rollover(state, events, rollover))
270    }
271
272    /// The history of an earlier (archived) run; empty if there is none.
273    pub fn run_history(&self, id: &str, run: u32) -> Result<Vec<HistoryEvent>> {
274        dispatch!(self, s => s.run_history(id, run))
275    }
276
277    /// The projected status of an execution (its latest run), if it exists.
278    pub fn execution_status(&self, id: &str) -> Result<Option<String>> {
279        dispatch!(self, s => s.execution_status(id))
280    }
281
282    /// True when this store cannot write (a cluster standby).
283    pub fn read_only(&self) -> bool {
284        matches!(self, Store::Postgres(p) if p.mode() == PgMode::ReadOnly)
285    }
286
287    /// True for a shared-mode store (several servers write).
288    pub fn shared(&self) -> bool {
289        matches!(self, Store::Postgres(p) if p.mode() == PgMode::Shared)
290    }
291
292    /// Drop every claim on a schedule occurrence. Only for a server that is
293    /// the only writer (single server, cluster leader): claims left behind
294    /// are from a process that is gone.
295    pub fn release_schedule_claims(&self) -> Result<u64> {
296        dispatch!(self, s => s.release_schedule_claims())
297    }
298
299    // -------------------------------------------------------------- schedules
300
301    /// Insert a new schedule; false if the id is taken.
302    pub fn insert_schedule(&self, row: &ScheduleRow) -> Result<bool> {
303        dispatch!(self, s => s.insert_schedule(row))
304    }
305
306    /// Overwrite a schedule's spec, paused flag and next occurrence, and drop
307    /// any claim on it; false if it does not exist.
308    pub fn update_schedule(&self, row: &ScheduleRow) -> Result<bool> {
309        dispatch!(self, s => s.update_schedule(row))
310    }
311
312    pub fn schedule(&self, id: &str) -> Result<Option<ScheduleRow>> {
313        dispatch!(self, s => s.schedule(id))
314    }
315
316    pub fn schedules(&self) -> Result<Vec<ScheduleRow>> {
317        dispatch!(self, s => s.schedules())
318    }
319
320    pub fn delete_schedule(&self, id: &str) -> Result<bool> {
321        dispatch!(self, s => s.delete_schedule(id))
322    }
323
324    /// Unpaused schedules whose next occurrence is due at `now` and that no
325    /// server holds a live claim on.
326    pub fn due_schedules(&self, now: u64) -> Result<Vec<ScheduleRow>> {
327        dispatch!(self, s => s.due_schedules(now))
328    }
329
330    /// Claim the occurrence `next_fire_ms` of a schedule for `node` until
331    /// `until`: a compare-and-set on the row, so of several servers that
332    /// find the same occurrence due exactly one gets it.
333    pub fn claim_schedule(
334        &self,
335        id: &str,
336        next_fire_ms: u64,
337        node: &str,
338        now: u64,
339        until: u64,
340    ) -> Result<bool> {
341        dispatch!(self, s => s.claim_schedule(id, next_fire_ms, node, now, until))
342    }
343
344    /// Record a handled occurrence and release the claim, if `node` still
345    /// holds it on occurrence `next_fire_ms`. False when the row changed
346    /// meanwhile (updated, paused, deleted or claimed by another server).
347    pub fn finish_schedule_fire(
348        &self,
349        id: &str,
350        next_fire_ms: u64,
351        node: &str,
352        now: u64,
353        record: &FireRecord,
354    ) -> Result<bool> {
355        dispatch!(self, s => s.finish_schedule_fire(id, next_fire_ms, node, now, record))
356    }
357
358    /// Give up a claim without advancing the schedule: the occurrence is
359    /// tried again from `retry_at` by any server.
360    pub fn release_schedule_claim(
361        &self,
362        id: &str,
363        node: &str,
364        retry_at: u64,
365        error: &str,
366        now: u64,
367    ) -> Result<()> {
368        dispatch!(self, s => s.release_schedule_claim(id, node, retry_at, error, now))
369    }
370
371    /// Note an execution started by `POST /v1/schedules/{id}/trigger`.
372    pub fn record_schedule_trigger(&self, id: &str, execution_id: &str, now: u64) -> Result<()> {
373        dispatch!(self, s => s.record_schedule_trigger(id, execution_id, now))
374    }
375
376    /// Ids of executions that have not reached a terminal status.
377    pub fn active_execution_ids(&self) -> Result<Vec<String>> {
378        dispatch!(self, s => s.active_execution_ids())
379    }
380
381    pub fn list_executions(
382        &self,
383        status: Option<&str>,
384        workflow: Option<&str>,
385        limit: u32,
386    ) -> Result<Vec<ExecutionSummary>> {
387        dispatch!(self, s => s.list_executions(status, workflow, limit))
388    }
389
390    pub fn count_by_status(&self) -> Result<Vec<(String, u64)>> {
391        dispatch!(self, s => s.count_by_status())
392    }
393
394    pub fn upsert_worker(&self, w: &WorkerRecord) -> Result<()> {
395        dispatch!(self, s => s.upsert_worker(w))
396    }
397
398    pub fn touch_worker(&self, worker_id: &str, now: u64) -> Result<()> {
399        dispatch!(self, s => s.touch_worker(worker_id, now))
400    }
401
402    pub fn workers(&self) -> Result<Vec<WorkerRecord>> {
403        dispatch!(self, s => s.workers())
404    }
405}
orbit_runtime/store/mod.rs

orbit_runtime/store/
mod.rs
