videre_core/
pipeline_runs.rs

1//! Per-command pipeline run history and liveness, surfaced by `videre stats`
2//! and other dashboard-style callers.
3//! See docs/superpowers/specs/2026-07-31-dashboard-stats-pass-b-design.md
4//! for the full design, and in particular why `track_in()` below does not rely
5//! on Drop/RAII for the success/failure bookkeeping: the library-scoped command
6//! lock's release does, backstopped by the OS releasing `flock` on any process
7//! death.
8
9use anyhow::{Context, Result};
10use rusqlite::{params, Connection, OptionalExtension};
11use serde::Serialize;
12
13/// The eight commands tracked so far. Extended with `locations` on
14/// 2026-08-01 (was seven after `prune`'s addition earlier the same day).
15/// It's a clean fit for the same one-shot start/finish model the others
16/// use: `videre locations` is a full-recompute batch pass, not an
17/// interactive per-query command. `gallery`, `search`, `mcp`, and `config`
18/// remain deliberately excluded: `report --faces`/`--show-faces` and `mcp`
19/// are long-running servers with no natural "finished" moment (the same
20/// reason `videre watch` itself is excluded. See below), `search` is an
21/// interactive per-query command rather than a library-processing pipeline
22/// stage (true even for its new `--location` mode, which is a single query
23/// like any other `search` invocation, not a batch job), and `config` is a
24/// trivial instant read/write with nothing meaningful to time. Revisit only
25/// if a real driver for tracking one of those emerges. See TECH_DEBT.md.
26///
27/// `videre watch` itself is deliberately not in this list, it has no
28/// "finished" moment during normal operation, so it gets its own liveness
29/// lock (see `watch_lock_path`) but no `pipeline_runs` row.
30pub const TRACKED_COMMANDS: [&str; 8] = [
31    "scan",
32    "faces",
33    "embed",
34    "classify",
35    "dedupe",
36    "fix-dates",
37    "prune",
38    "locations",
39];
40
41pub fn ensure_pipeline_runs_table(conn: &Connection) -> rusqlite::Result<()> {
42    conn.execute_batch(
43        "CREATE TABLE IF NOT EXISTS pipeline_runs (
44            command      TEXT PRIMARY KEY,
45            started_at   TEXT NOT NULL,
46            finished_at  TEXT,
47            status       TEXT NOT NULL,
48            duration_ms  INTEGER,
49            summary      TEXT
50        );",
51    )
52}
53
54pub fn start_run(conn: &Connection, command: &str) -> rusqlite::Result<()> {
55    conn.execute(
56        "INSERT INTO pipeline_runs (command, started_at, status)
57         VALUES (?1, datetime('now'), 'running')
58         ON CONFLICT(command) DO UPDATE SET
59             started_at = excluded.started_at,
60             status = 'running',
61             finished_at = NULL,
62             duration_ms = NULL,
63             summary = NULL",
64        params![command],
65    )?;
66    Ok(())
67}
68
69pub fn finish_run(
70    conn: &Connection,
71    command: &str,
72    status: &str,
73    duration_ms: i64,
74    summary: Option<&str>,
75) -> rusqlite::Result<()> {
76    conn.execute(
77        "UPDATE pipeline_runs SET
78             finished_at = datetime('now'),
79             status = ?2,
80             duration_ms = ?3,
81             summary = ?4
82         WHERE command = ?1",
83        params![command, status, duration_ms, summary],
84    )?;
85    Ok(())
86}
87
88/// Wraps `f` with pipeline-run bookkeeping for a command whose per-command
89/// lock is already held as a
90/// [`library_locks::CommandGuard`](crate::library_locks::CommandGuard).
91///
92/// The guard is verified rather than trusted: it must have been taken for
93/// exactly `ctx` and `command`, because a mismatched pairing would record
94/// one library's run against another (or one command's row against another
95/// command's) with nothing visibly wrong. The model is `start_run` before
96/// `f`, `success`/`failed` (with the error message) after, all before
97/// returning. It never reacquires the supplied guard, so converting a command
98/// to the library-scoped locks cannot double-lock.
99pub fn track_in<T, F>(
100    conn: &Connection,
101    ctx: &crate::library::LibraryContext,
102    guard: &crate::library_locks::CommandGuard,
103    command: &str,
104    f: F,
105) -> Result<T>
106where
107    F: FnOnce() -> Result<T>,
108{
109    guard.ensure_matches(ctx, command)?;
110    // The guard's match is a wiring check on path strings; this rechecks that
111    // the root still names the same library before any run row is written, so
112    // a root swapped out between guard acquisition and now is refused rather
113    // than recorded against whatever now sits at the path.
114    ctx.ensure_root_identity()?;
115    ensure_pipeline_runs_table(conn)?;
116    start_run(conn, command)?;
117    let started = std::time::Instant::now();
118    let result = f();
119    let duration_ms = started.elapsed().as_millis().min(i64::MAX as u128) as i64;
120    match result {
121        Ok(value) => {
122            finish_run(conn, command, "success", duration_ms, None)?;
123            Ok(value)
124        }
125        Err(error) => {
126            // Record the failure, but never let a bookkeeping error mask the
127            // real one: the original error is what the caller acted on.
128            if let Err(record_error) = finish_run(
129                conn,
130                command,
131                "failed",
132                duration_ms,
133                Some(&error.to_string()),
134            ) {
135                return Err(error.context(format!(
136                    "also could not record the failed run: {record_error}"
137                )));
138            }
139            Err(error)
140        }
141    }
142}
143
144/// Every tracked command's last run and current liveness. Liveness is probed
145/// through the library's own lock files (`library_locks::command_locked`), so a
146/// library-scoped command shows up as running exactly when a library-scoped
147/// reader asks; a `running` row whose lock no live process holds reads back as
148/// `crashed`.
149pub fn read_all_in(
150    conn: &Connection,
151    ctx: &crate::library::LibraryContext,
152) -> Result<Vec<PipelineRunStatus>> {
153    ensure_pipeline_runs_table(conn)?;
154    let mut out = Vec::with_capacity(TRACKED_COMMANDS.len());
155    for command in TRACKED_COMMANDS {
156        let row: Option<(String, Option<i64>, String)> = conn
157            .query_row(
158                "SELECT started_at, duration_ms, status FROM pipeline_runs WHERE command = ?1",
159                params![command],
160                |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)),
161            )
162            .optional()?;
163        let currently_running = crate::library_locks::command_locked(ctx, command)?;
164        let (last_run_at, status, duration_ms) = match row {
165            None => (None, None, None),
166            Some((started_at, duration_ms, stored_status)) => {
167                let status = if stored_status == "running" && !currently_running {
168                    "crashed".to_string()
169                } else {
170                    stored_status
171                };
172                (Some(started_at), Some(status), duration_ms)
173            }
174        };
175        out.push(PipelineRunStatus {
176            command: command.to_string(),
177            last_run_at,
178            status,
179            duration_ms,
180            currently_running,
181        });
182    }
183    Ok(out)
184}
185
186/// Install a SIGINT handler that marks `command`'s row `interrupted` and exits
187/// 130, against the library the context pins.
188///
189/// The library's identity is validated before the handler is installed (and
190/// again inside it): a handler bound to a root that no longer names its
191/// library would write another library's database on the way out. The
192/// handler's connection opens without `CREATE`, so an exiting process can
193/// never conjure a database that initialization is responsible for. Same
194/// best-effort contract as the global version: errors inside the handler
195/// are swallowed, there is no useful way to report them once the process is
196/// already exiting on a signal.
197pub fn install_sigint_handler_in(
198    ctx: std::sync::Arc<crate::library::LibraryContext>,
199    command: &'static str,
200) -> Result<()> {
201    ctx.ensure_root_identity()
202        .context("validating the library before installing the SIGINT handler")?;
203    ctrlc::set_handler(move || {
204        if ctx.ensure_root_identity().is_ok() {
205            if let Ok(conn) = crate::library_db::open_without_create(&ctx.paths.db) {
206                let _ = conn.busy_timeout(std::time::Duration::from_secs(5));
207                let started_at: Option<String> = conn
208                    .query_row(
209                        "SELECT started_at FROM pipeline_runs WHERE command = ?1",
210                        params![command],
211                        |r| r.get(0),
212                    )
213                    .optional()
214                    .ok()
215                    .flatten();
216                let duration_ms = started_at
217                    .and_then(|s| {
218                        chrono::NaiveDateTime::parse_from_str(&s, "%Y-%m-%d %H:%M:%S").ok()
219                    })
220                    .map(|started| {
221                        (chrono::Utc::now().naive_utc() - started)
222                            .num_milliseconds()
223                            .max(0)
224                    })
225                    .unwrap_or(0);
226                let _ = finish_run(&conn, command, "interrupted", duration_ms, None);
227            }
228        }
229        std::process::exit(130);
230    })
231    .context("installing SIGINT handler")
232}
233
234#[derive(Debug, Clone, PartialEq, Serialize)]
235pub struct PipelineRunStatus {
236    pub command: String,
237    pub last_run_at: Option<String>,
238    /// "running" | "success" | "failed" | "interrupted" | "crashed" | None if never run.
239    /// "crashed" is computed here, never a stored value. See the design doc.
240    pub status: Option<String>,
241    pub duration_ms: Option<i64>,
242    pub currently_running: bool,
243}
244
245#[cfg(test)]
246mod tests {
247    use super::*;
248
249    fn test_db() -> Connection {
250        let conn = Connection::open_in_memory().unwrap();
251        ensure_pipeline_runs_table(&conn).unwrap();
252        conn
253    }
254
255    #[test]
256    fn ensure_pipeline_runs_table_is_idempotent() {
257        let conn = test_db();
258        ensure_pipeline_runs_table(&conn).unwrap();
259    }
260
261    #[test]
262    fn start_run_then_finish_run_records_success() {
263        let conn = test_db();
264        start_run(&conn, "embed").unwrap();
265
266        let status: String = conn
267            .query_row(
268                "SELECT status FROM pipeline_runs WHERE command = 'embed'",
269                [],
270                |r| r.get(0),
271            )
272            .unwrap();
273        assert_eq!(status, "running");
274
275        finish_run(&conn, "embed", "success", 1234, None).unwrap();
276
277        let (status, duration_ms, summary): (String, i64, Option<String>) = conn
278            .query_row(
279                "SELECT status, duration_ms, summary FROM pipeline_runs WHERE command = 'embed'",
280                [],
281                |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)),
282            )
283            .unwrap();
284        assert_eq!(status, "success");
285        assert_eq!(duration_ms, 1234);
286        assert_eq!(summary, None);
287    }
288
289    #[test]
290    fn start_run_upserts_resetting_prior_finish_fields() {
291        let conn = test_db();
292        start_run(&conn, "embed").unwrap();
293        finish_run(&conn, "embed", "failed", 500, Some("boom")).unwrap();
294
295        start_run(&conn, "embed").unwrap(); // second run begins
296
297        let (status, duration_ms, summary): (String, Option<i64>, Option<String>) = conn
298            .query_row(
299                "SELECT status, duration_ms, summary FROM pipeline_runs WHERE command = 'embed'",
300                [],
301                |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)),
302            )
303            .unwrap();
304        assert_eq!(status, "running");
305        assert_eq!(duration_ms, None);
306        assert_eq!(summary, None);
307
308        let count: i64 = conn
309            .query_row("SELECT COUNT(*) FROM pipeline_runs", [], |r| r.get(0))
310            .unwrap();
311        assert_eq!(count, 1, "upsert, not a second row");
312    }
313
314    /// One library with its state and locks directories in place, plus a
315    /// connection with the runs table: the minimum the library-scoped
316    /// bookkeeping needs, without pulling the database layer's
317    /// initialization in. Locks only need the directories to exist.
318    fn in_library() -> (
319        tempfile::TempDir,
320        crate::library::LibraryContext,
321        Connection,
322    ) {
323        let temp = tempfile::tempdir().unwrap();
324        let root = temp.path().join("photos");
325        std::fs::create_dir(&root).unwrap();
326        let ctx = crate::library::LibraryContext::new(&root, &temp.path().join("cache")).unwrap();
327        std::fs::create_dir_all(&ctx.paths.locks).unwrap();
328        let conn = Connection::open_in_memory().unwrap();
329        ensure_pipeline_runs_table(&conn).unwrap();
330        (temp, ctx, conn)
331    }
332
333    #[test]
334    fn track_in_records_runs_under_an_already_held_command_guard() {
335        let (_t, ctx, conn) = in_library();
336        let guard = crate::library_locks::try_command(&ctx, "scan").unwrap();
337        let result = track_in(&conn, &ctx, &guard, "scan", || Ok(7)).unwrap();
338        assert_eq!(result, 7);
339        let (status, summary): (String, Option<String>) = conn
340            .query_row(
341                "SELECT status, summary FROM pipeline_runs WHERE command = 'scan'",
342                [],
343                |r| Ok((r.get(0)?, r.get(1)?)),
344            )
345            .unwrap();
346        assert_eq!(status, "success");
347        assert_eq!(summary, None);
348        // The other commands' rows are untouched: one command's bookkeeping
349        // never overwrites another's.
350        let count: i64 = conn
351            .query_row("SELECT COUNT(*) FROM pipeline_runs", [], |r| r.get(0))
352            .unwrap();
353        assert_eq!(count, 1);
354        let failed: Result<()> =
355            track_in(&conn, &ctx, &guard, "scan", || Err(anyhow::anyhow!("boom")));
356        failed.unwrap_err();
357        let (status, summary): (String, Option<String>) = conn
358            .query_row(
359                "SELECT status, summary FROM pipeline_runs WHERE command = 'scan'",
360                [],
361                |r| Ok((r.get(0)?, r.get(1)?)),
362            )
363            .unwrap();
364        assert_eq!(status, "failed");
365        assert_eq!(summary.as_deref(), Some("boom"));
366    }
367
368    #[test]
369    fn track_in_refuses_a_guard_from_another_command_or_library() {
370        let (_t, ctx, conn) = in_library();
371        let guard = crate::library_locks::try_command(&ctx, "scan").unwrap();
372        // A guard held for scan must not bookkeep embed's row.
373        let result: Result<()> = track_in(&conn, &ctx, &guard, "embed", || Ok(()));
374        let err = result.unwrap_err();
375        assert!(format!("{err:#}").contains("scan"), "{err:#}");
376        let count: i64 = conn
377            .query_row("SELECT COUNT(*) FROM pipeline_runs", [], |r| r.get(0))
378            .unwrap();
379        assert_eq!(count, 0, "a refused guard must write no row");
380
381        // Nor may one library's guard bookkeep another library's run, even
382        // under the same command name.
383        let temp = tempfile::tempdir().unwrap();
384        let other_root = temp.path().join("other");
385        std::fs::create_dir(&other_root).unwrap();
386        let other =
387            crate::library::LibraryContext::new(&other_root, &temp.path().join("cache")).unwrap();
388        let result: Result<()> = track_in(&conn, &other, &guard, "scan", || Ok(()));
389        let err = result.unwrap_err();
390        assert!(
391            format!("{err:#}").contains(other_root.file_name().unwrap().to_string_lossy().as_ref()),
392            "{err:#}"
393        );
394        let count: i64 = conn
395            .query_row("SELECT COUNT(*) FROM pipeline_runs", [], |r| r.get(0))
396            .unwrap();
397        assert_eq!(count, 0);
398    }
399
400    #[test]
401    fn read_all_in_answers_liveness_from_the_library_locks() {
402        let (_t, ctx, conn) = in_library();
403        let guard = crate::library_locks::try_command(&ctx, "scan").unwrap();
404        track_in(&conn, &ctx, &guard, "scan", || Ok(())).unwrap();
405        // Released: the finished run is not running.
406        drop(guard);
407        // A held command lock shows up as running to a library-scoped
408        // reader, with no row of its own: the probe is what says live, not
409        // the presence of a row.
410        let _faces = crate::library_locks::try_command(&ctx, "faces").unwrap();
411        let statuses = read_all_in(&conn, &ctx).unwrap();
412        let scan = statuses.iter().find(|s| s.command == "scan").unwrap();
413        assert_eq!(scan.status.as_deref(), Some("success"));
414        assert!(!scan.currently_running);
415        let faces = statuses.iter().find(|s| s.command == "faces").unwrap();
416        assert_eq!(faces.status, None);
417        assert!(faces.currently_running);
418    }
419
420    #[test]
421    fn install_sigint_handler_in_validates_the_library_before_installing() {
422        // A context whose root was replaced after construction must be
423        // refused before any handler is installed, so this deliberately
424        // never reaches ctrlc::set_handler (only one handler can exist per
425        // process, and another test in this suite owns that slot).
426        let temp = tempfile::tempdir().unwrap();
427        let root = temp.path().join("photos");
428        std::fs::create_dir(&root).unwrap();
429        let ctx = std::sync::Arc::new(
430            crate::library::LibraryContext::new(&root, &temp.path().join("cache")).unwrap(),
431        );
432        std::fs::rename(&root, temp.path().join("moved")).unwrap();
433        std::fs::create_dir(&root).unwrap();
434        let err = install_sigint_handler_in(ctx, "scan").unwrap_err();
435        assert!(format!("{err:#}").contains("no longer names"), "{err:#}");
436    }
437}
videre_core/pipeline_runs.rs

videre_core/
pipeline_runs.rs
