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

faucet_cli/catalog/
mod.rs
