lash_sqlite_store/process_registry/
support.rs

1use super::*;
2
3pub(super) fn process_status_label(record: &ProcessRecord) -> &'static str {
4    record.status.label()
5}
6
7impl SqliteProcessRegistry {
8    pub async fn open(path: &Path) -> tokio_rusqlite::Result<Self> {
9        Self::open_with_clock(path, Arc::new(lash_core::SystemClock)).await
10    }
11
12    pub async fn open_with_clock(
13        path: &Path,
14        clock: Arc<dyn lash_core::Clock>,
15    ) -> tokio_rusqlite::Result<Self> {
16        let conn = SqliteConnection::open(path).await?;
17        ensure_process_schema(&conn).await?;
18        apply_pragmas(&conn, StoreBacking::File).await?;
19        Ok(Self { conn, clock })
20    }
21
22    pub async fn memory() -> tokio_rusqlite::Result<Self> {
23        Self::memory_with_clock(Arc::new(lash_core::SystemClock)).await
24    }
25
26    pub async fn memory_with_clock(
27        clock: Arc<dyn lash_core::Clock>,
28    ) -> tokio_rusqlite::Result<Self> {
29        let conn = SqliteConnection::open_in_memory().await?;
30        ensure_process_schema(&conn).await?;
31        apply_pragmas(&conn, StoreBacking::Memory).await?;
32        Ok(Self { conn, clock })
33    }
34
35    pub(crate) fn load_process_conn(
36        conn: &Connection,
37        process_id: &str,
38    ) -> Result<Option<ProcessRecord>, lash_core::PluginError> {
39        let json: Option<String> = conn
40            .query_row(
41                "SELECT record_json FROM processes WHERE process_id = ?1",
42                params![process_id],
43                |row| row.get(0),
44            )
45            .optional()
46            .map_err(process_sqlite_error)?;
47        json.map(|json| serde_json::from_str(&json).map_err(process_decode_error))
48            .transpose()
49    }
50
51    pub(crate) fn save_process_conn(
52        conn: &Connection,
53        record: &ProcessRecord,
54    ) -> Result<(), lash_core::PluginError> {
55        let change_seq = Self::next_change_seq_conn(conn)?;
56        conn.execute(
57            "UPDATE processes
58             SET updated_at_ms = ?2, change_seq = ?3, status = ?4, record_json = ?5
59             WHERE process_id = ?1",
60            params![
61                record.id.as_str(),
62                record.updated_at_ms as i64,
63                change_seq as i64,
64                process_status_label(record),
65                process_encode_json(record)?
66            ],
67        )
68        .map_err(process_sqlite_error)?;
69        Ok(())
70    }
71
72    pub(super) fn next_change_seq_conn(conn: &Connection) -> Result<u64, lash_core::PluginError> {
73        conn.execute(
74            "UPDATE process_change_clock
75             SET current_seq = current_seq + 1
76             WHERE singleton = 1",
77            [],
78        )
79        .map_err(process_sqlite_error)?;
80        conn.query_row(
81            "SELECT current_seq FROM process_change_clock WHERE singleton = 1",
82            [],
83            |row| row.get::<_, i64>(0),
84        )
85        .map(|seq| seq as u64)
86        .map_err(process_sqlite_error)
87    }
88
89    pub(crate) fn load_event_by_key_conn(
90        conn: &Connection,
91        process_id: &str,
92        replay_key: &str,
93    ) -> Result<Option<(String, ProcessEvent)>, lash_core::PluginError> {
94        let row: Option<(String, String)> = conn
95            .query_row(
96                "SELECT payload_hash, event_json
97                 FROM process_events
98                 WHERE process_id = ?1 AND idempotency_key = ?2",
99                params![process_id, replay_key],
100                |row| Ok((row.get(0)?, row.get(1)?)),
101            )
102            .optional()
103            .map_err(process_sqlite_error)?;
104        row.map(|(hash, json)| {
105            serde_json::from_str(&json)
106                .map(|event| (hash, event))
107                .map_err(process_decode_error)
108        })
109        .transpose()
110    }
111
112    pub(crate) fn load_process_lease_conn(
113        conn: &Connection,
114        process_id: &str,
115    ) -> Result<Option<ProcessLease>, lash_core::PluginError> {
116        conn.query_row(
117            "SELECT lease_owner_id, lease_token, lease_fencing_token,
118                    lease_claimed_at_ms, lease_expires_at_ms,
119                    lease_owner_incarnation_id, lease_owner_liveness_json
120             FROM process_leases
121             WHERE process_id = ?1",
122            params![process_id],
123            |row| {
124                let owner_id: Option<String> = row.get(0)?;
125                let lease_token: Option<String> = row.get(1)?;
126                let incarnation_id: Option<String> = row.get(5)?;
127                let liveness_json: Option<String> = row.get(6)?;
128                let (Some(owner_id), Some(lease_token)) = (owner_id, lease_token) else {
129                    return Ok(None);
130                };
131                Ok(Some(ProcessLease {
132                    schema_version: PROCESS_LEASE_SCHEMA_VERSION,
133                    process_id: process_id.to_string(),
134                    owner: process_lease_owner_from_columns(
135                        owner_id,
136                        incarnation_id,
137                        liveness_json,
138                    ),
139                    lease_token,
140                    fencing_token: row.get::<_, i64>(2)? as u64,
141                    claimed_at_epoch_ms: row.get::<_, i64>(3)? as u64,
142                    expires_at_epoch_ms: row.get::<_, i64>(4)? as u64,
143                }))
144            },
145        )
146        .optional()
147        .map(|lease| lease.flatten())
148        .map_err(process_sqlite_error)
149    }
150
151    /// Insert-or-replace the persisted lease row for `process_id` with a fresh
152    /// lease owned by `owner` at `fencing_token`.
153    pub(super) fn acquire_process_lease_conn(
154        conn: &Connection,
155        process_id: &str,
156        owner: &LeaseOwnerIdentity,
157        fencing_token: u64,
158        now: u64,
159        lease_ttl_ms: u64,
160    ) -> Result<ProcessLease, lash_core::PluginError> {
161        let lease = ProcessLease {
162            schema_version: PROCESS_LEASE_SCHEMA_VERSION,
163            process_id: process_id.to_string(),
164            owner: owner.clone(),
165            lease_token: format!(
166                "{:x}",
167                Sha256::digest(
168                    format!(
169                        "{process_id}:{}:{}:{now}:{fencing_token}",
170                        owner.owner_id, owner.incarnation_id
171                    )
172                    .as_bytes()
173                )
174            ),
175            fencing_token,
176            claimed_at_epoch_ms: now,
177            expires_at_epoch_ms: now.saturating_add(lease_ttl_ms),
178        };
179        conn.execute(
180            "INSERT INTO process_leases (
181                process_id, lease_owner_id, lease_owner_incarnation_id,
182                lease_owner_liveness_json, lease_token, lease_fencing_token,
183                lease_claimed_at_ms, lease_expires_at_ms
184             )
185             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)
186             ON CONFLICT(process_id) DO UPDATE SET
187                lease_owner_id = excluded.lease_owner_id,
188                lease_owner_incarnation_id = excluded.lease_owner_incarnation_id,
189                lease_owner_liveness_json = excluded.lease_owner_liveness_json,
190                lease_token = excluded.lease_token,
191                lease_fencing_token = excluded.lease_fencing_token,
192                lease_claimed_at_ms = excluded.lease_claimed_at_ms,
193                lease_expires_at_ms = excluded.lease_expires_at_ms",
194            params![
195                lease.process_id.as_str(),
196                lease.owner.owner_id.as_str(),
197                lease.owner.incarnation_id.as_str(),
198                encode_process_lease_liveness(&lease.owner.liveness)?,
199                lease.lease_token.as_str(),
200                lease.fencing_token as i64,
201                lease.claimed_at_epoch_ms as i64,
202                lease.expires_at_epoch_ms as i64,
203            ],
204        )
205        .map_err(process_sqlite_error)?;
206        Ok(lease)
207    }
208
209    pub(super) fn list_grants_for_scope_conn(
210        conn: &Connection,
211        session_scope: &SessionScope,
212        live_only: bool,
213    ) -> Result<Vec<ProcessHandleGrantEntry>, lash_core::PluginError> {
214        let session_scope_id = session_scope.id();
215        let status_clause = if live_only {
216            "AND p.status = 'running'"
217        } else {
218            ""
219        };
220        let mut stmt = conn
221            .prepare(&format!(
222                "SELECT g.process_id, g.descriptor_json, p.record_json
223                 FROM process_handle_grants g
224                 JOIN processes p ON p.process_id = g.process_id
225                 WHERE g.scope_id = ?1 {status_clause}
226                 ORDER BY g.process_id ASC"
227            ))
228            .map_err(process_sqlite_error)?;
229        let rows = stmt
230            .query_map(params![session_scope_id.as_str()], |row| {
231                Ok((
232                    row.get::<_, String>(0)?,
233                    row.get::<_, String>(1)?,
234                    row.get::<_, String>(2)?,
235                ))
236            })
237            .map_err(process_sqlite_error)?;
238        let mut entries = Vec::new();
239        for row in rows {
240            let (process_id, descriptor_json, record_json) = row.map_err(process_sqlite_error)?;
241            let descriptor: ProcessHandleDescriptor =
242                serde_json::from_str(&descriptor_json).map_err(process_decode_error)?;
243            let record: ProcessRecord =
244                serde_json::from_str(&record_json).map_err(process_decode_error)?;
245            entries.push((
246                ProcessHandleGrant {
247                    session_id: session_scope.session_id.clone(),
248                    process_id,
249                    descriptor,
250                },
251                record,
252            ));
253        }
254        Ok(entries)
255    }
256}
257
258/// Map a `Result<T, PluginError>` produced by a synchronous transaction body to
259/// a [`TxOutcome`]: commit on success, roll back on logical error. Both arms
260/// carry the inner `Result` back so the caller recovers the value or the
261/// `PluginError` after the transaction resolves.
262pub(crate) fn tx_outcome<T>(
263    result: Result<T, lash_core::PluginError>,
264) -> TxOutcome<Result<T, lash_core::PluginError>> {
265    match result {
266        Ok(value) => TxOutcome::Commit(Ok(value)),
267        Err(err) => TxOutcome::Rollback(Err(err)),
268    }
269}
lash_sqlite_store/process_registry/support.rs

lash_sqlite_store/process_registry/
support.rs
