orbit_runtime/store/
sqlite.rs

1//! SQLite backend (default; a single file, WAL mode with `synchronous=FULL`).
2
3use super::{ExecutionSummary, PutDefinition, Result, WorkerRecord, SCHEMA};
4use crate::model::{Event, ExecutionState, HistoryEvent};
5use orbit_ir::WorkflowIr;
6use rusqlite::{params, Connection, OptionalExtension};
7use std::path::Path;
8
9pub struct SqliteStore {
10    conn: Connection,
11}
12
13impl SqliteStore {
14    pub fn open(path: &Path) -> Result<SqliteStore> {
15        let conn = Connection::open(path)?;
16        Self::init(conn)
17    }
18
19    pub fn in_memory() -> Result<SqliteStore> {
20        Self::init(Connection::open_in_memory()?)
21    }
22
23    fn init(conn: Connection) -> Result<SqliteStore> {
24        // WAL + FULL sync: a committed event survives power loss.
25        conn.pragma_update(None, "journal_mode", "WAL")?;
26        conn.pragma_update(None, "synchronous", "FULL")?;
27        conn.pragma_update(None, "foreign_keys", "ON")?;
28        conn.busy_timeout(std::time::Duration::from_secs(5))?;
29        conn.execute_batch(&SCHEMA.replace("BIGINT", "INTEGER"))?;
30        if conn.query_row("SELECT COUNT(*) FROM schema_version", [], |r| {
31            r.get::<_, i64>(0)
32        })? == 0
33        {
34            conn.execute("INSERT INTO schema_version (version) VALUES (1)", [])?;
35        }
36        Ok(SqliteStore { conn })
37    }
38
39    // ------------------------------------------------------------ definitions
40
41    pub fn put_definition(&self, ir: &WorkflowIr, now: u64) -> Result<PutDefinition> {
42        let existing: Option<String> = self
43            .conn
44            .query_row(
45                "SELECT definition_hash FROM workflow_definitions WHERE name = ?1 AND version = ?2",
46                params![ir.name, ir.version],
47                |r| r.get(0),
48            )
49            .optional()?;
50        match existing {
51            Some(h) if h == ir.definition_hash => Ok(PutDefinition::Unchanged),
52            Some(h) => Ok(PutDefinition::Conflict { existing_hash: h }),
53            None => {
54                let text = serde_json::to_string(ir).expect("IR serialises");
55                self.conn.execute(
56                    "INSERT INTO workflow_definitions (name, version, definition_hash, ir, created_at_ms) VALUES (?1, ?2, ?3, ?4, ?5)",
57                    params![ir.name, ir.version, ir.definition_hash, text, now as i64],
58                )?;
59                Ok(PutDefinition::Created)
60            }
61        }
62    }
63
64    pub fn definitions(&self) -> Result<Vec<WorkflowIr>> {
65        let mut stmt = self
66            .conn
67            .prepare("SELECT ir FROM workflow_definitions ORDER BY name, created_at_ms")?;
68        let rows = stmt.query_map([], |r| r.get::<_, String>(0))?;
69        let mut out = vec![];
70        for text in rows {
71            match serde_json::from_str::<WorkflowIr>(&text?) {
72                Ok(ir) => out.push(ir),
73                Err(e) => tracing::error!(error = %e, "skipping unreadable workflow definition"),
74            }
75        }
76        Ok(out)
77    }
78
79    // ------------------------------------------------------------- executions
80
81    pub fn execution_exists(&self, id: &str) -> Result<bool> {
82        Ok(self
83            .conn
84            .query_row(
85                "SELECT 1 FROM workflow_executions WHERE execution_id = ?1",
86                params![id],
87                |_| Ok(()),
88            )
89            .optional()?
90            .is_some())
91    }
92
93    /// Atomically append events and update the execution projection.
94    pub fn append(&mut self, state: &ExecutionState, events: &[HistoryEvent]) -> Result<()> {
95        let tx = self.conn.transaction()?;
96        {
97            let mut ins = tx.prepare_cached(
98                "INSERT INTO execution_events (execution_id, seq, event_type, step, payload, timestamp_ms) VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
99            )?;
100            for h in events {
101                let payload = serde_json::to_string(&h.event).expect("event serialises");
102                ins.execute(params![
103                    state.id,
104                    h.seq as i64,
105                    h.event.name(),
106                    h.event.step(),
107                    payload,
108                    h.timestamp_ms as i64
109                ])?;
110            }
111        }
112        tx.execute(
113            "INSERT INTO workflow_executions (execution_id, workflow, version, definition_hash, status, created_at_ms, updated_at_ms, last_seq)
114             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)
115             ON CONFLICT(execution_id) DO UPDATE SET status = excluded.status, updated_at_ms = excluded.updated_at_ms, last_seq = excluded.last_seq",
116            params![
117                state.id,
118                state.workflow,
119                state.version,
120                state.definition_hash,
121                state.status.as_str(),
122                state.created_at_ms as i64,
123                state.updated_at_ms as i64,
124                state.last_seq as i64
125            ],
126        )?;
127        Ok(tx.commit()?)
128    }
129
130    pub fn history(&self, id: &str) -> Result<Vec<HistoryEvent>> {
131        let mut stmt = self.conn.prepare_cached(
132            "SELECT seq, timestamp_ms, payload FROM execution_events WHERE execution_id = ?1 ORDER BY seq",
133        )?;
134        let rows = stmt.query_map(params![id], |r| {
135            Ok((
136                r.get::<_, i64>(0)?,
137                r.get::<_, i64>(1)?,
138                r.get::<_, String>(2)?,
139            ))
140        })?;
141        let mut out = vec![];
142        for row in rows {
143            let (seq, ts, payload) = row?;
144            let event: Event = serde_json::from_str(&payload).map_err(|e| {
145                rusqlite::Error::FromSqlConversionFailure(
146                    2,
147                    rusqlite::types::Type::Text,
148                    Box::new(e),
149                )
150            })?;
151            out.push(HistoryEvent {
152                seq: seq as u64,
153                timestamp_ms: ts as u64,
154                event,
155            });
156        }
157        Ok(out)
158    }
159
160    /// Ids of executions that have not reached a terminal status.
161    pub fn active_execution_ids(&self) -> Result<Vec<String>> {
162        let mut stmt = self.conn.prepare(
163            "SELECT execution_id FROM workflow_executions WHERE status IN ('RUNNING', 'COMPENSATING') ORDER BY created_at_ms",
164        )?;
165        let rows = stmt.query_map([], |r| r.get(0))?;
166        Ok(rows.collect::<std::result::Result<Vec<_>, _>>()?)
167    }
168
169    pub fn list_executions(
170        &self,
171        status: Option<&str>,
172        workflow: Option<&str>,
173        limit: u32,
174    ) -> Result<Vec<ExecutionSummary>> {
175        let mut stmt = self.conn.prepare(
176            "SELECT execution_id, workflow, version, status, created_at_ms, updated_at_ms FROM workflow_executions
177             WHERE (?1 IS NULL OR status = ?1) AND (?2 IS NULL OR workflow = ?2)
178             ORDER BY created_at_ms DESC LIMIT ?3",
179        )?;
180        let rows = stmt.query_map(params![status, workflow, limit], |r| {
181            Ok(ExecutionSummary {
182                execution_id: r.get(0)?,
183                workflow: r.get(1)?,
184                version: r.get(2)?,
185                status: r.get(3)?,
186                created_at_ms: r.get::<_, i64>(4)? as u64,
187                updated_at_ms: r.get::<_, i64>(5)? as u64,
188            })
189        })?;
190        Ok(rows.collect::<std::result::Result<Vec<_>, _>>()?)
191    }
192
193    pub fn count_by_status(&self) -> Result<Vec<(String, u64)>> {
194        let mut stmt = self
195            .conn
196            .prepare("SELECT status, COUNT(*) FROM workflow_executions GROUP BY status")?;
197        let rows = stmt.query_map([], |r| Ok((r.get(0)?, r.get::<_, i64>(1)? as u64)))?;
198        Ok(rows.collect::<std::result::Result<Vec<_>, _>>()?)
199    }
200
201    // ---------------------------------------------------------------- workers
202
203    pub fn upsert_worker(&self, w: &WorkerRecord) -> Result<()> {
204        self.conn.execute(
205            "INSERT INTO worker_registry (worker_id, runtime, sdk, contracts, registered_at_ms, last_seen_ms) VALUES (?1, ?2, ?3, ?4, ?5, ?6)
206             ON CONFLICT(worker_id) DO UPDATE SET runtime = excluded.runtime, sdk = excluded.sdk, contracts = excluded.contracts,
207             registered_at_ms = excluded.registered_at_ms, last_seen_ms = excluded.last_seen_ms",
208            params![w.worker_id, w.runtime, w.sdk, w.contracts.to_string(), w.registered_at_ms as i64, w.last_seen_ms as i64],
209        )?;
210        Ok(())
211    }
212
213    pub fn touch_worker(&self, worker_id: &str, now: u64) -> Result<()> {
214        self.conn.execute(
215            "UPDATE worker_registry SET last_seen_ms = ?2 WHERE worker_id = ?1",
216            params![worker_id, now as i64],
217        )?;
218        Ok(())
219    }
220
221    pub fn workers(&self) -> Result<Vec<WorkerRecord>> {
222        let mut stmt = self.conn.prepare(
223            "SELECT worker_id, runtime, sdk, contracts, registered_at_ms, last_seen_ms FROM worker_registry ORDER BY worker_id",
224        )?;
225        let rows = stmt.query_map([], |r| {
226            Ok(WorkerRecord {
227                worker_id: r.get(0)?,
228                runtime: r.get(1)?,
229                sdk: r.get(2)?,
230                contracts: serde_json::from_str(&r.get::<_, String>(3)?).unwrap_or_default(),
231                registered_at_ms: r.get::<_, i64>(4)? as u64,
232                last_seen_ms: r.get::<_, i64>(5)? as u64,
233            })
234        })?;
235        Ok(rows.collect::<std::result::Result<Vec<_>, _>>()?)
236    }
237}
orbit_runtime/store/sqlite.rs

orbit_runtime/store/
sqlite.rs
