faucet_cli/catalog/
mod.rs

1//! Data Movement Catalog (#279) — the CLI-side write path.
2//!
3//! The catalog accumulates, run over run, the operational history of every
4//! dataset a pipeline touches: identity, a deduplicated schema timeline,
5//! volume/freshness stats, and lineage edges. Storage rides the serve
6//! run-history backends (`crate::serve::history`); this module is the glue
7//! the executor calls after every successful **root** invocation:
8//!
9//! - [`spec`] — the top-level `catalog:` config block (`faucet schema catalog`).
10//! - [`model`] — pure URI canonicalization + sample-schema inference.
11//! - [`CatalogHandle`] / [`connect_from_spec`] — the store handle carried on
12//!   `ExecuteOptions` (serve passes its own history backend; the CLI runtimes
13//!   connect from the `catalog:` block).
14//! - [`record`] — the never-fails-the-run write (mirrors the SLA / lineage
15//!   "log-and-continue" contract).
16//!
17//! Gated on the `catalog` Cargo feature (implies `serve` for the storage
18//! backends and `lineage` for record sampling + column-lineage derivation).
19
20pub mod model;
21pub mod snapshot;
22pub mod spec;
23
24pub use spec::CatalogSpec;
25
26use crate::error::{CliError, CliResult};
27use crate::serve::config::HistoryBackendSpec;
28use crate::serve::history::catalog::ConfigSnapshot;
29use crate::serve::history::{self, RunHistory, catalog::CatalogUpdate};
30use std::sync::Arc;
31use std::time::Duration;
32
33/// Default per-side schema-inference sample cap when no `catalog:` block set
34/// one (the serve write path, which has no block).
35pub const DEFAULT_SAMPLE_RECORDS: usize = 100;
36
37/// The catalog store handle carried on `ExecuteOptions`. Cheaply cloneable.
38#[derive(Clone)]
39pub struct CatalogHandle {
40    pub store: Arc<dyn RunHistory>,
41    /// Provenance run id recorded on every catalog row this run produces —
42    /// the serve run id when running under `faucet serve`, `None` for CLI
43    /// runtimes (each invocation then stamps its own observability run id).
44    pub run_id: Option<String>,
45    /// Per-side schema-inference sample cap.
46    pub sample_records: usize,
47}
48
49impl std::fmt::Debug for CatalogHandle {
50    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
51        f.debug_struct("CatalogHandle")
52            .field("run_id", &self.run_id)
53            .field("sample_records", &self.sample_records)
54            .finish_non_exhaustive()
55    }
56}
57
58/// Build a catalog store from the `catalog:` block. Errors are config-level
59/// (bad scheme / missing build feature) and fail fast at load time; an
60/// *unreachable* SQL backend degrades to in-memory via `FallbackHistory`
61/// (logged, run unaffected) exactly like `faucet serve --history`.
62pub async fn connect_from_spec(spec: &CatalogSpec) -> CliResult<CatalogHandle> {
63    let backend = parse_url(&spec.url)?;
64    let store = history::connect(
65        &backend,
66        // Idempotency claims + run leases are run-history concerns; the
67        // catalog-only connection never uses them.
68        Duration::from_secs(3600),
69        Duration::from_secs(30),
70        &uuid::Uuid::now_v7().to_string(),
71    )
72    .await?;
73    Ok(CatalogHandle {
74        store,
75        run_id: None,
76        sample_records: spec.sample_records,
77    })
78}
79
80/// Parse the `catalog.url` field into a history-backend selection.
81fn parse_url(url: &str) -> CliResult<HistoryBackendSpec> {
82    match url {
83        "memory" => Ok(HistoryBackendSpec::Memory),
84        u if u.starts_with("postgres://") || u.starts_with("postgresql://") => {
85            Ok(HistoryBackendSpec::Postgres(u.to_string()))
86        }
87        u if u.starts_with("sqlite:") => Ok(HistoryBackendSpec::Sqlite(u.to_string())),
88        other => Err(CliError::Config(format!(
89            "catalog.url '{other}' is not recognised — expected 'memory', 'sqlite:<path>', \
90             or a 'postgres://…' URL"
91        ))),
92    }
93}
94
95/// Persist one run's catalog update. Monitoring must never take down the run
96/// it observes: any backend error is logged once per call and swallowed.
97pub async fn record(handle: &CatalogHandle, update: &CatalogUpdate) {
98    if let Err(e) = handle.store.catalog_record(update).await {
99        tracing::warn!(
100            pipeline = %update.pipeline,
101            row = %update.row,
102            error = %e,
103            "catalog write failed — run unaffected"
104        );
105    }
106}
107
108/// Persist the latest resolved+expanded config snapshot for `faucet plan --diff`
109/// (#374). Best-effort, same never-fails-the-run contract as [`record`].
110pub async fn record_config_snapshot(handle: &CatalogHandle, snapshot: &ConfigSnapshot) {
111    if let Err(e) = handle.store.catalog_record_config_snapshot(snapshot).await {
112        tracing::warn!(
113            pipeline = %snapshot.pipeline,
114            error = %e,
115            "config-snapshot write failed — run unaffected"
116        );
117    }
118}
119
120#[cfg(test)]
121mod tests {
122    use super::*;
123
124    #[tokio::test]
125    async fn connect_memory_and_reject_unknown_scheme() {
126        let handle = connect_from_spec(&CatalogSpec {
127            url: "memory".into(),
128            sample_records: 25,
129        })
130        .await
131        .unwrap();
132        assert_eq!(handle.sample_records, 25);
133        assert!(handle.run_id.is_none());
134
135        let err = connect_from_spec(&CatalogSpec {
136            url: "mysql://nope".into(),
137            sample_records: 100,
138        })
139        .await
140        .unwrap_err();
141        assert!(err.to_string().contains("catalog.url"), "{err}");
142    }
143
144    #[test]
145    fn parse_url_recognises_all_three_schemes() {
146        assert!(matches!(
147            parse_url("sqlite:./cat.db"),
148            Ok(HistoryBackendSpec::Sqlite(u)) if u == "sqlite:./cat.db"
149        ));
150        assert!(matches!(
151            parse_url("postgres://h/db"),
152            Ok(HistoryBackendSpec::Postgres(_))
153        ));
154        assert!(matches!(
155            parse_url("postgresql://h/db"),
156            Ok(HistoryBackendSpec::Postgres(_))
157        ));
158        assert!(matches!(
159            parse_url("memory"),
160            Ok(HistoryBackendSpec::Memory)
161        ));
162        assert!(parse_url("bogus").is_err());
163    }
164
165    #[tokio::test]
166    async fn handle_debug_never_prints_the_store() {
167        let handle = connect_from_spec(&CatalogSpec {
168            url: "memory".into(),
169            sample_records: 7,
170        })
171        .await
172        .unwrap();
173        let dbg = format!("{handle:?}");
174        assert!(dbg.contains("sample_records: 7"), "{dbg}");
175        assert!(dbg.contains(".."), "non-exhaustive marker expected: {dbg}");
176    }
177}
faucet_cli/catalog/mod.rs

faucet_cli/catalog/
mod.rs
