cuttlefish_host/ledger.rs
1//! Per-job durable checkpoint store. One SQLite file per job
2//! (`$CUTTLEFISH_HOME/jobs/<job_id>/ledger.sqlite`), matching the catalog's
3//! existing one-thing-per-file convention. See
4//! docs/superpowers/specs/2026-08-03-dag-core-design.md's "Durability model"
5//! for the full rationale — this module is purely storage; the resume
6//! decision logic (skip on completed/skipped, run everything else) lives in
7//! `crate::runner`.
8
9use rusqlite::Connection;
10use std::path::Path;
11use std::sync::Mutex;
12
13/// A job's own terminal status, as recorded in the ledger — distinct from
14/// per-node checkpoints, which alone can't tell "still running when the
15/// process died" apart from "finished cleanly."
16#[derive(Debug, Clone, Copy, PartialEq, Eq)]
17pub enum LedgerJobStatus {
18 /// The job has not yet called [`Ledger::finish`].
19 Running,
20 /// The job finished successfully.
21 Completed,
22 /// The job finished with an error.
23 Failed,
24 /// The job was cancelled before it finished.
25 Cancelled,
26}
27
28impl LedgerJobStatus {
29 fn from_str(s: &str) -> Self {
30 match s {
31 "completed" => Self::Completed,
32 "failed" => Self::Failed,
33 "cancelled" => Self::Cancelled,
34 _ => Self::Running,
35 }
36 }
37}
38
39/// A per-job checkpoint ledger, backed by a single SQLite file.
40///
41/// The connection is behind a `Mutex` purely to make `Ledger: Sync` —
42/// `rusqlite::Connection` is `Send` but not `Sync` (its statement cache uses
43/// unsynchronized interior mutability), and `run_job` holds a `&Ledger`
44/// across `.await` points in a task spawned onto a multi-threaded runtime,
45/// which requires the held reference to be `Send`, which in turn requires
46/// `Ledger: Sync`. There is normally only ever one writer (the job that owns
47/// this ledger), so contention is not a real concern; the lock exists to
48/// satisfy the type system's threading rules, not to arbitrate real
49/// concurrent access.
50pub struct Ledger {
51 conn: Mutex<Connection>,
52}
53
54impl Ledger {
55 /// Open (creating if absent) a job's ledger at `path`, ensuring both
56 /// tables exist and `job_status` has its single row.
57 ///
58 /// `graph_fingerprint` is recorded only when `job_status` doesn't exist
59 /// yet (a fresh job) — reopening an existing ledger leaves the
60 /// originally-recorded fingerprint untouched, since comparing old vs.
61 /// new fingerprint is a resume endpoint's job, not `open`'s.
62 pub fn open(path: &Path, graph_fingerprint: &str) -> rusqlite::Result<Self> {
63 if let Some(parent) = path.parent() {
64 std::fs::create_dir_all(parent).ok();
65 }
66 let conn = Connection::open(path)?;
67 // A future reader (daemon startup scan, resume endpoint) may open
68 // its own connection to this same file while a running job's
69 // connection is mid-write. Without this, SQLite's default
70 // busy_timeout of 0 means that second connection gets an immediate
71 // SQLITE_BUSY instead of waiting briefly for the lock to clear.
72 conn.busy_timeout(std::time::Duration::from_secs(5))?;
73 conn.execute_batch(
74 "CREATE TABLE IF NOT EXISTS checkpoints (
75 node_name TEXT PRIMARY KEY,
76 status TEXT NOT NULL,
77 output_json TEXT,
78 completed_at TEXT NOT NULL
79 );
80 CREATE TABLE IF NOT EXISTS job_status (status TEXT NOT NULL, graph_fingerprint TEXT NOT NULL);",
81 )?;
82 let count: i64 = conn.query_row("SELECT COUNT(*) FROM job_status", [], |r| r.get(0))?;
83 if count == 0 {
84 conn.execute(
85 "INSERT INTO job_status (status, graph_fingerprint) VALUES ('running', ?1)",
86 [graph_fingerprint],
87 )?;
88 }
89 Ok(Self {
90 conn: Mutex::new(conn),
91 })
92 }
93
94 /// The fingerprint recorded when this job was first submitted — compare
95 /// against a freshly computed one before resuming.
96 pub fn graph_fingerprint(&self) -> rusqlite::Result<String> {
97 self.lock()
98 .query_row("SELECT graph_fingerprint FROM job_status", [], |r| r.get(0))
99 }
100
101 /// The recorded output of `node_name`, if it completed successfully.
102 /// `None` for a node that never ran, is still pending, or was skipped.
103 pub fn get_completed(&self, node_name: &str) -> rusqlite::Result<Option<serde_json::Value>> {
104 let result: Option<(String, Option<String>)> = match self.lock().query_row(
105 "SELECT status, output_json FROM checkpoints WHERE node_name = ?1",
106 [node_name],
107 |r| Ok((r.get(0)?, r.get(1)?)),
108 ) {
109 Ok(row) => Some(row),
110 Err(rusqlite::Error::QueryReturnedNoRows) => None,
111 Err(e) => return Err(e),
112 };
113 match result {
114 Some((status, Some(json))) if status == "completed" => Ok(Some(
115 serde_json::from_str(&json).expect("ledger never stores invalid JSON"),
116 )),
117 _ => Ok(None),
118 }
119 }
120
121 /// Whether `node_name` was recorded as skipped (e.g. excluded by a
122 /// branch decision).
123 pub fn is_skipped(&self, node_name: &str) -> rusqlite::Result<bool> {
124 let status: Option<String> = match self.lock().query_row(
125 "SELECT status FROM checkpoints WHERE node_name = ?1",
126 [node_name],
127 |r| r.get(0),
128 ) {
129 Ok(status) => Some(status),
130 Err(rusqlite::Error::QueryReturnedNoRows) => None,
131 Err(e) => return Err(e),
132 };
133 Ok(status.as_deref() == Some("skipped"))
134 }
135
136 /// Record `node_name` as completed with `output`, overwriting any prior
137 /// checkpoint for that node.
138 pub fn write_completed(
139 &self,
140 node_name: &str,
141 output: &serde_json::Value,
142 ) -> rusqlite::Result<()> {
143 self.lock().execute(
144 "INSERT OR REPLACE INTO checkpoints (node_name, status, output_json, completed_at)
145 VALUES (?1, 'completed', ?2, ?3)",
146 rusqlite::params![node_name, output.to_string(), now_marker()],
147 )?;
148 Ok(())
149 }
150
151 /// Record `node_name` as skipped, overwriting any prior checkpoint for
152 /// that node.
153 pub fn write_skipped(&self, node_name: &str) -> rusqlite::Result<()> {
154 self.lock().execute(
155 "INSERT OR REPLACE INTO checkpoints (node_name, status, output_json, completed_at)
156 VALUES (?1, 'skipped', NULL, ?2)",
157 rusqlite::params![node_name, now_marker()],
158 )?;
159 Ok(())
160 }
161
162 /// The job's own terminal status. `Running` until [`Ledger::finish`] is
163 /// called.
164 pub fn job_status(&self) -> rusqlite::Result<LedgerJobStatus> {
165 let s: String = self
166 .lock()
167 .query_row("SELECT status FROM job_status", [], |r| r.get(0))?;
168 Ok(LedgerJobStatus::from_str(&s))
169 }
170
171 /// Record the job's terminal status (e.g. `"completed"`, `"failed"`,
172 /// `"cancelled"`).
173 pub fn finish(&self, status: &str) -> rusqlite::Result<()> {
174 self.lock()
175 .execute("UPDATE job_status SET status = ?1", [status])?;
176 Ok(())
177 }
178
179 /// Lock the connection. The mutex is only ever held for the duration of
180 /// one synchronous rusqlite call, never across an `.await` — so a
181 /// poisoned lock can only mean a prior call panicked mid-query, an
182 /// exceptional situation worth propagating loudly rather than papering
183 /// over.
184 fn lock(&self) -> std::sync::MutexGuard<'_, Connection> {
185 self.conn.lock().expect("ledger connection mutex poisoned")
186 }
187}
188
189/// The root directory jobs live under. Checks `$CUTTLEFISH_JOBS_HOME` first
190/// — set by `cuttlefish-run`'s project-scoping so a project's jobs/ledger
191/// state lives under `<project>/.cuttlefish/jobs` without also redirecting
192/// the (deliberately still-global) block catalog, which `$CUTTLEFISH_HOME`
193/// alone continues to control. Falls back to `$CUTTLEFISH_HOME/jobs` (or
194/// `~/.cuttlefish/jobs`) exactly as before when unset, so nothing about
195/// existing single-global-home behavior changes for a caller that never
196/// sets the new variable.
197pub fn jobs_root() -> Option<std::path::PathBuf> {
198 if let Ok(dir) = std::env::var("CUTTLEFISH_JOBS_HOME") {
199 return Some(std::path::PathBuf::from(dir));
200 }
201 crate::catalog::cuttlefish_home().map(|h| h.join("jobs"))
202}
203
204/// A completed_at marker. Plain wall-clock formatting, same as the
205/// catalog's `now_rfc3339` (`crate::catalog`) — this is a diagnostic field,
206/// not consulted by any resume logic, so precision/format choices here
207/// don't affect correctness.
208fn now_marker() -> String {
209 crate::catalog::now_rfc3339()
210}