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/// Records that a recurring command (`watch`) completed one cycle. Unlike a
145/// start/stop run, the heartbeat is the whole record: `started_at` is the
146/// moment of the last successful cycle, and there is no duration to keep. A
147/// watcher that dies mid-cycle leaves the previous cycle's heartbeat, so a
148/// dead watcher reads as a stale last-cycle time, never as a fake crash.
149pub fn record_heartbeat_in(
150    conn: &Connection,
151    ctx: &crate::library::LibraryContext,
152    command: &str,
153) -> Result<()> {
154    // Same identity check as `track_in`: a root swapped between the cycle's
155    // start and now must not have its cycle recorded against another library.
156    ctx.ensure_root_identity()?;
157    ensure_pipeline_runs_table(conn)?;
158    conn.execute(
159        "INSERT INTO pipeline_runs (command, started_at, status, summary)
160         VALUES (?1, datetime('now'), 'success', 'last successful cycle')
161         ON CONFLICT(command) DO UPDATE SET
162             started_at = excluded.started_at,
163             status = 'success',
164             finished_at = NULL,
165             duration_ms = NULL,
166             summary = excluded.summary",
167        params![command],
168    )?;
169    Ok(())
170}
171
172/// Every tracked command's last run and current liveness. Liveness is probed
173/// through the library's own lock files (`library_locks::command_locked`), so a
174/// library-scoped command shows up as running exactly when a library-scoped
175/// reader asks; a `running` row whose lock no live process holds reads back as
176/// `crashed`.
177pub fn read_all_in(
178    conn: &Connection,
179    ctx: &crate::library::LibraryContext,
180) -> Result<Vec<PipelineRunStatus>> {
181    ensure_pipeline_runs_table(conn)?;
182    let mut out = Vec::with_capacity(TRACKED_COMMANDS.len());
183    for command in TRACKED_COMMANDS {
184        out.push(read_one_in(conn, ctx, command)?);
185    }
186    // `watch` is deliberately absent from TRACKED_COMMANDS: it is a recurring
187    // loop, not a start/stop run, so it never takes the per-command run-row
188    // machinery. Its heartbeat (see `record_heartbeat_in`) makes it reportable
189    // all the same; it appears here only once a heartbeat exists, so a library
190    // never watched is not lectured about a stage it never started.
191    if conn
192        .query_row(
193            "SELECT 1 FROM pipeline_runs WHERE command = 'watch'",
194            [],
195            |r| r.get::<_, i64>(0),
196        )
197        .optional()?
198        .is_some()
199    {
200        out.push(read_one_in(conn, ctx, "watch")?);
201    }
202    Ok(out)
203}
204
205/// One command's row plus lock-derived liveness. Shared by the tracked loop
206/// and the watch heartbeat read, which need identical semantics.
207fn read_one_in(
208    conn: &Connection,
209    ctx: &crate::library::LibraryContext,
210    command: &str,
211) -> Result<PipelineRunStatus> {
212    let row: Option<(String, Option<i64>, String)> = conn
213        .query_row(
214            "SELECT started_at, duration_ms, status FROM pipeline_runs WHERE command = ?1",
215            params![command],
216            |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)),
217        )
218        .optional()?;
219    let currently_running = crate::library_locks::command_locked(ctx, command)?;
220    let (last_run_at, status, duration_ms) = match row {
221        None => (None, None, None),
222        Some((started_at, duration_ms, stored_status)) => {
223            let status = if stored_status == "running" && !currently_running {
224                "crashed".to_string()
225            } else {
226                stored_status
227            };
228            (Some(started_at), Some(status), duration_ms)
229        }
230    };
231    Ok(PipelineRunStatus {
232        command: command.to_string(),
233        last_run_at,
234        status,
235        duration_ms,
236        currently_running,
237    })
238}
239
240/// Install a SIGINT handler that marks `command`'s row `interrupted` and exits
241/// 130, against the library the context pins.
242///
243/// The library's identity is validated before the handler is installed (and
244/// again inside it): a handler bound to a root that no longer names its
245/// library would write another library's database on the way out. The
246/// handler's connection opens without `CREATE`, so an exiting process can
247/// never conjure a database that initialization is responsible for. Same
248/// best-effort contract as the global version: errors inside the handler
249/// are swallowed, there is no useful way to report them once the process is
250/// already exiting on a signal.
251/// A `videre pipeline` runs several tracked stages in one process, but the
252/// `ctrlc` crate permits exactly one handler per process. So the handler is
253/// installed once and then retargeted: each stage records the command it is
254/// running, and the single handler marks whichever command is current when the
255/// signal arrives. Re-installing is a silent no-op rather than the "already
256/// registered" error every stage after the first used to raise.
257static SIGINT_INSTALLED: std::sync::atomic::AtomicBool = std::sync::atomic::AtomicBool::new(false);
258static SIGINT_COMMAND: std::sync::Mutex<Option<&'static str>> = std::sync::Mutex::new(None);
259
260pub fn install_sigint_handler_in(
261    ctx: std::sync::Arc<crate::library::LibraryContext>,
262    command: &'static str,
263) -> Result<()> {
264    ctx.ensure_root_identity()
265        .context("validating the library before installing the SIGINT handler")?;
266    // Point the single, process-wide handler at the stage now running.
267    if let Ok(mut current) = SIGINT_COMMAND.lock() {
268        *current = Some(command);
269    }
270    // Later stages in the same process only retarget the handler above; one
271    // handler per process is all `ctrlc` allows.
272    if SIGINT_INSTALLED.load(std::sync::atomic::Ordering::SeqCst) {
273        return Ok(());
274    }
275    ctrlc::set_handler(move || {
276        let command = SIGINT_COMMAND
277            .lock()
278            .ok()
279            .and_then(|c| *c)
280            .unwrap_or(command);
281        if ctx.ensure_root_identity().is_ok() {
282            if let Ok(conn) = crate::library_db::open_without_create(&ctx.paths.db) {
283                let _ = conn.busy_timeout(std::time::Duration::from_secs(5));
284                let started_at: Option<String> = conn
285                    .query_row(
286                        "SELECT started_at FROM pipeline_runs WHERE command = ?1",
287                        params![command],
288                        |r| r.get(0),
289                    )
290                    .optional()
291                    .ok()
292                    .flatten();
293                let duration_ms = started_at
294                    .and_then(|s| {
295                        chrono::NaiveDateTime::parse_from_str(&s, "%Y-%m-%d %H:%M:%S").ok()
296                    })
297                    .map(|started| {
298                        (chrono::Utc::now().naive_utc() - started)
299                            .num_milliseconds()
300                            .max(0)
301                    })
302                    .unwrap_or(0);
303                let _ = finish_run(&conn, command, "interrupted", duration_ms, None);
304            }
305        }
306        std::process::exit(130);
307    })
308    .context("installing SIGINT handler")?;
309    SIGINT_INSTALLED.store(true, std::sync::atomic::Ordering::SeqCst);
310    Ok(())
311}
312
313#[derive(Debug, Clone, PartialEq, Serialize)]
314pub struct PipelineRunStatus {
315    pub command: String,
316    pub last_run_at: Option<String>,
317    /// "running" | "success" | "failed" | "interrupted" | "crashed" | None if never run.
318    /// "crashed" is computed here, never a stored value. See the design doc.
319    pub status: Option<String>,
320    pub duration_ms: Option<i64>,
321    pub currently_running: bool,
322}
323
324#[cfg(test)]
325mod tests {
326    use super::*;
327
328    fn test_db() -> Connection {
329        let conn = Connection::open_in_memory().unwrap();
330        ensure_pipeline_runs_table(&conn).unwrap();
331        conn
332    }
333
334    #[test]
335    fn heartbeat_records_last_cycle_and_reads_back() {
336        let (_t, ctx, conn) = in_library();
337        // A fresh library has never run watch: no row, no liveness time.
338        assert!(read_all_in(&conn, &ctx)
339            .unwrap()
340            .iter()
341            .all(|r| r.command != "watch"));
342        record_heartbeat_in(&conn, &ctx, "watch").unwrap();
343        let w = read_all_in(&conn, &ctx)
344            .unwrap()
345            .into_iter()
346            .find(|r| r.command == "watch")
347            .expect("the heartbeat row must surface in the run read");
348        assert!(w.last_run_at.is_some(), "started_at is the last-cycle time");
349        assert_eq!(w.status.as_deref(), Some("success"));
350        assert!(!w.currently_running, "no watch process holds the lock");
351
352        // A second heartbeat is an update of the same row, not an error or a
353        // second entry: recurring commands have exactly one now.
354        record_heartbeat_in(&conn, &ctx, "watch").unwrap();
355        assert_eq!(
356            read_all_in(&conn, &ctx)
357                .unwrap()
358                .iter()
359                .filter(|r| r.command == "watch")
360                .count(),
361            1
362        );
363    }
364
365    #[test]
366    fn ensure_pipeline_runs_table_is_idempotent() {
367        let conn = test_db();
368        ensure_pipeline_runs_table(&conn).unwrap();
369    }
370
371    #[test]
372    fn start_run_then_finish_run_records_success() {
373        let conn = test_db();
374        start_run(&conn, "embed").unwrap();
375
376        let status: String = conn
377            .query_row(
378                "SELECT status FROM pipeline_runs WHERE command = 'embed'",
379                [],
380                |r| r.get(0),
381            )
382            .unwrap();
383        assert_eq!(status, "running");
384
385        finish_run(&conn, "embed", "success", 1234, None).unwrap();
386
387        let (status, duration_ms, summary): (String, i64, Option<String>) = conn
388            .query_row(
389                "SELECT status, duration_ms, summary FROM pipeline_runs WHERE command = 'embed'",
390                [],
391                |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)),
392            )
393            .unwrap();
394        assert_eq!(status, "success");
395        assert_eq!(duration_ms, 1234);
396        assert_eq!(summary, None);
397    }
398
399    #[test]
400    fn start_run_upserts_resetting_prior_finish_fields() {
401        let conn = test_db();
402        start_run(&conn, "embed").unwrap();
403        finish_run(&conn, "embed", "failed", 500, Some("boom")).unwrap();
404
405        start_run(&conn, "embed").unwrap(); // second run begins
406
407        let (status, duration_ms, summary): (String, Option<i64>, Option<String>) = conn
408            .query_row(
409                "SELECT status, duration_ms, summary FROM pipeline_runs WHERE command = 'embed'",
410                [],
411                |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)),
412            )
413            .unwrap();
414        assert_eq!(status, "running");
415        assert_eq!(duration_ms, None);
416        assert_eq!(summary, None);
417
418        let count: i64 = conn
419            .query_row("SELECT COUNT(*) FROM pipeline_runs", [], |r| r.get(0))
420            .unwrap();
421        assert_eq!(count, 1, "upsert, not a second row");
422    }
423
424    /// One library with its state and locks directories in place, plus a
425    /// connection with the runs table: the minimum the library-scoped
426    /// bookkeeping needs, without pulling the database layer's
427    /// initialization in. Locks only need the directories to exist.
428    fn in_library() -> (
429        tempfile::TempDir,
430        crate::library::LibraryContext,
431        Connection,
432    ) {
433        let temp = tempfile::tempdir().unwrap();
434        let root = temp.path().join("photos");
435        std::fs::create_dir(&root).unwrap();
436        let ctx = crate::library::LibraryContext::new(&root, &temp.path().join("cache")).unwrap();
437        std::fs::create_dir_all(&ctx.paths.locks).unwrap();
438        let conn = Connection::open_in_memory().unwrap();
439        ensure_pipeline_runs_table(&conn).unwrap();
440        (temp, ctx, conn)
441    }
442
443    #[test]
444    fn track_in_records_runs_under_an_already_held_command_guard() {
445        let (_t, ctx, conn) = in_library();
446        let guard = crate::library_locks::try_command(&ctx, "scan").unwrap();
447        let result = track_in(&conn, &ctx, &guard, "scan", || Ok(7)).unwrap();
448        assert_eq!(result, 7);
449        let (status, summary): (String, Option<String>) = conn
450            .query_row(
451                "SELECT status, summary FROM pipeline_runs WHERE command = 'scan'",
452                [],
453                |r| Ok((r.get(0)?, r.get(1)?)),
454            )
455            .unwrap();
456        assert_eq!(status, "success");
457        assert_eq!(summary, None);
458        // The other commands' rows are untouched: one command's bookkeeping
459        // never overwrites another's.
460        let count: i64 = conn
461            .query_row("SELECT COUNT(*) FROM pipeline_runs", [], |r| r.get(0))
462            .unwrap();
463        assert_eq!(count, 1);
464        let failed: Result<()> =
465            track_in(&conn, &ctx, &guard, "scan", || Err(anyhow::anyhow!("boom")));
466        failed.unwrap_err();
467        let (status, summary): (String, Option<String>) = conn
468            .query_row(
469                "SELECT status, summary FROM pipeline_runs WHERE command = 'scan'",
470                [],
471                |r| Ok((r.get(0)?, r.get(1)?)),
472            )
473            .unwrap();
474        assert_eq!(status, "failed");
475        assert_eq!(summary.as_deref(), Some("boom"));
476    }
477
478    #[test]
479    fn track_in_refuses_a_guard_from_another_command_or_library() {
480        let (_t, ctx, conn) = in_library();
481        let guard = crate::library_locks::try_command(&ctx, "scan").unwrap();
482        // A guard held for scan must not bookkeep embed's row.
483        let result: Result<()> = track_in(&conn, &ctx, &guard, "embed", || Ok(()));
484        let err = result.unwrap_err();
485        assert!(format!("{err:#}").contains("scan"), "{err:#}");
486        let count: i64 = conn
487            .query_row("SELECT COUNT(*) FROM pipeline_runs", [], |r| r.get(0))
488            .unwrap();
489        assert_eq!(count, 0, "a refused guard must write no row");
490
491        // Nor may one library's guard bookkeep another library's run, even
492        // under the same command name.
493        let temp = tempfile::tempdir().unwrap();
494        let other_root = temp.path().join("other");
495        std::fs::create_dir(&other_root).unwrap();
496        let other =
497            crate::library::LibraryContext::new(&other_root, &temp.path().join("cache")).unwrap();
498        let result: Result<()> = track_in(&conn, &other, &guard, "scan", || Ok(()));
499        let err = result.unwrap_err();
500        assert!(
501            format!("{err:#}").contains(other_root.file_name().unwrap().to_string_lossy().as_ref()),
502            "{err:#}"
503        );
504        let count: i64 = conn
505            .query_row("SELECT COUNT(*) FROM pipeline_runs", [], |r| r.get(0))
506            .unwrap();
507        assert_eq!(count, 0);
508    }
509
510    #[test]
511    fn read_all_in_answers_liveness_from_the_library_locks() {
512        let (_t, ctx, conn) = in_library();
513        let guard = crate::library_locks::try_command(&ctx, "scan").unwrap();
514        track_in(&conn, &ctx, &guard, "scan", || Ok(())).unwrap();
515        // Released: the finished run is not running.
516        drop(guard);
517        // A held command lock shows up as running to a library-scoped
518        // reader, with no row of its own: the probe is what says live, not
519        // the presence of a row.
520        let _faces = crate::library_locks::try_command(&ctx, "faces").unwrap();
521        let statuses = read_all_in(&conn, &ctx).unwrap();
522        let scan = statuses.iter().find(|s| s.command == "scan").unwrap();
523        assert_eq!(scan.status.as_deref(), Some("success"));
524        assert!(!scan.currently_running);
525        let faces = statuses.iter().find(|s| s.command == "faces").unwrap();
526        assert_eq!(faces.status, None);
527        assert!(faces.currently_running);
528    }
529
530    #[test]
531    fn install_sigint_handler_in_validates_the_library_before_installing() {
532        // A context whose root was replaced after construction must be
533        // refused before any handler is installed, so this deliberately
534        // never reaches ctrlc::set_handler (only one handler can exist per
535        // process, and another test in this suite owns that slot).
536        let temp = tempfile::tempdir().unwrap();
537        let root = temp.path().join("photos");
538        std::fs::create_dir(&root).unwrap();
539        let ctx = std::sync::Arc::new(
540            crate::library::LibraryContext::new(&root, &temp.path().join("cache")).unwrap(),
541        );
542        std::fs::rename(&root, temp.path().join("moved")).unwrap();
543        std::fs::create_dir(&root).unwrap();
544        let err = install_sigint_handler_in(ctx, "scan").unwrap_err();
545        assert!(format!("{err:#}").contains("no longer names"), "{err:#}");
546    }
547
548    #[test]
549    fn install_sigint_handler_in_is_idempotent_within_a_process() {
550        // A pipeline installs the handler once and retargets it for each later
551        // stage, so a second install must succeed silently rather than raise
552        // "already registered" - the warning pipeline used to print between
553        // scan and faces.
554        let temp = tempfile::tempdir().unwrap();
555        let root = temp.path().join("photos");
556        std::fs::create_dir(&root).unwrap();
557        let ctx = std::sync::Arc::new(
558            crate::library::LibraryContext::new(&root, &temp.path().join("cache")).unwrap(),
559        );
560        install_sigint_handler_in(ctx.clone(), "scan").expect("first install");
561        install_sigint_handler_in(ctx, "faces")
562            .expect("a second install in the same process must be a no-op, not an error");
563    }
564}
videre_core/pipeline_runs.rs

videre_core/
pipeline_runs.rs
