faucet_core/
rollback.rs

1//! Undo a run (#706): the shared types behind `faucet rollback`.
2//!
3//! A run that should not have happened — a wrong parameter, a corrupt upstream
4//! extract, a bad deploy — is reverted by the sink that wrote it, one dataset at
5//! a time, through the defaulted [`Sink::rollback_run`](crate::Sink::rollback_run)
6//! hook. What "revert" means depends on how the run wrote:
7//!
8//! - **append** — delete the rows stamped with the run id (the `_faucet_run_id`
9//!   metadata column, which the `rollback:` block turns on);
10//! - **upsert / delete** — deleting is not enough, because the run overwrote
11//!   older values; with the opt-in **before-image journal**
12//!   ([`RUN_JOURNAL_TABLE`]) the sink recorded the prior row of every key it
13//!   changed, in the same transaction as the page, and rollback restores it;
14//! - **overwrite** — with `keep_previous`, the replaced table is kept as
15//!   `<table>__faucet_prev` ([`PREVIOUS_TABLE_SUFFIX`]) and rollback swaps it back.
16//!
17//! The pipeline's state bookmark and exactly-once commit token are rewound by
18//! the CLI from the pre-run marker it recorded, so the next run re-reads what
19//! the reverted run consumed. Everything here is pure data; the sinks and the
20//! CLI carry the I/O.
21
22use serde::{Deserialize, Serialize};
23
24/// The before-image journal table a journaling sink maintains next to its
25/// data tables: `(run_id, table_name, key_json, before_json)`; a `NULL`
26/// before-image means the key did not exist before the run.
27pub const RUN_JOURNAL_TABLE: &str = "_faucet_run_journal";
28
29/// Suffix of the table an overwrite run keeps its replaced contents in when
30/// `keep_previous` is on.
31pub const PREVIOUS_TABLE_SUFFIX: &str = "__faucet_prev";
32
33/// Default name of the run-id column rollback keys on
34/// (`{metadata prefix}_{run_id}` with the default prefix).
35pub const DEFAULT_RUN_ID_COLUMN: &str = "_faucet_run_id";
36
37/// Runtime rollback settings the executor hands a sink through its
38/// [`WriteSpec`](crate::write_mode::WriteSpec) (`rollback:` field). Not user
39/// config: the top-level `rollback:` block decides these, and the CLI injects
40/// them per invocation.
41#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
42#[serde(deny_unknown_fields)]
43pub struct RollbackWriteSpec {
44    /// The run id this invocation writes under.
45    pub run_id: String,
46    /// The destination column carrying the run id (stamped by the metadata
47    /// decorator).
48    pub run_id_column: String,
49    /// Record before-images of every upserted / deleted key.
50    #[serde(default)]
51    pub journal: bool,
52    /// Keep the replaced table of an overwrite as `<table>__faucet_prev`.
53    #[serde(default)]
54    pub keep_previous: bool,
55}
56
57/// How the run being rolled back wrote, which decides the undo strategy.
58#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
59#[serde(rename_all = "snake_case")]
60pub enum RollbackMode {
61    /// Delete the run's rows by run id.
62    Append,
63    /// Restore before-images from the journal.
64    Upsert,
65    /// Swap the kept previous table back.
66    Overwrite,
67}
68
69impl RollbackMode {
70    /// The undo strategy for a write mode.
71    pub fn for_write_mode(mode: crate::write_mode::WriteMode) -> Self {
72        match mode {
73            crate::write_mode::WriteMode::Append => RollbackMode::Append,
74            crate::write_mode::WriteMode::Upsert | crate::write_mode::WriteMode::Delete => {
75                RollbackMode::Upsert
76            }
77            crate::write_mode::WriteMode::Overwrite => RollbackMode::Overwrite,
78        }
79    }
80
81    pub fn as_str(self) -> &'static str {
82        match self {
83            RollbackMode::Append => "append",
84            RollbackMode::Upsert => "upsert",
85            RollbackMode::Overwrite => "overwrite",
86        }
87    }
88}
89
90/// What a sink is asked to undo.
91#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
92pub struct RollbackOptions {
93    /// The destination column carrying the run id.
94    pub run_id_column: String,
95    pub mode: RollbackMode,
96    /// Restore a key even when a later run changed it since (the row's run id
97    /// no longer matches). Off by default: such a key is a **conflict** and the
98    /// whole dataset is left untouched.
99    pub force: bool,
100    /// Count what would change without changing anything.
101    pub dry_run: bool,
102}
103
104/// What a sink did (or would do) for one dataset.
105#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
106pub struct RollbackOutcome {
107    /// Rows deleted (append: the run's rows; upsert: keys the run created).
108    pub deleted: u64,
109    /// Rows restored to their before-image (upsert), or rows swapped back
110    /// (overwrite).
111    pub restored: u64,
112    /// Keys a later run changed since; without `force` these block the whole
113    /// dataset.
114    pub conflicts: u64,
115    /// Whether changes were applied. `false` for a dry run, or when conflicts
116    /// blocked the rollback.
117    pub applied: bool,
118    /// A human note (e.g. why nothing was applied).
119    #[serde(default, skip_serializing_if = "Option::is_none")]
120    pub note: Option<String>,
121}
122
123impl RollbackOutcome {
124    /// Nothing to undo: the run left no rows / journal / previous table.
125    pub fn nothing(note: impl Into<String>) -> Self {
126        Self {
127            applied: true,
128            note: Some(note.into()),
129            ..Default::default()
130        }
131    }
132
133    /// Blocked by conflicts (no change made).
134    pub fn blocked(conflicts: u64) -> Self {
135        Self {
136            conflicts,
137            applied: false,
138            note: Some(format!(
139                "{conflicts} key(s) were changed by a later run; pass --force to restore them anyway"
140            )),
141            ..Default::default()
142        }
143    }
144}
145
146/// The canonical JSON text of a key tuple — the journal's row identity.
147/// Columns are emitted in the configured key order, so the same key always
148/// serialises identically.
149pub fn key_json(tuple: &crate::write_mode::KeyTuple) -> String {
150    // Built by hand rather than through a `Map`: the map's key order depends
151    // on the `preserve_order` feature, and this text is a primary key.
152    let parts: Vec<String> = tuple
153        .0
154        .iter()
155        .map(|(k, v)| format!("{}:{}", serde_json::Value::String(k.clone()), v))
156        .collect();
157    format!("{{{}}}", parts.join(","))
158}
159
160/// A key tuple with every scalar rendered as its text form, so a key read
161/// back from the database (`7`) and the same key in a source record (`"7"`)
162/// journal identically. Nulls stay null; the SQL sinks bind the text through
163/// the key column's own type, so the restore addresses the right row.
164pub fn canonical_key(tuple: &crate::write_mode::KeyTuple) -> crate::write_mode::KeyTuple {
165    crate::write_mode::KeyTuple(
166        tuple
167            .0
168            .iter()
169            .map(|(k, v)| {
170                let v = match v {
171                    serde_json::Value::Null => serde_json::Value::Null,
172                    serde_json::Value::String(s) => serde_json::Value::String(s.clone()),
173                    serde_json::Value::Bool(b) => serde_json::Value::String(b.to_string()),
174                    serde_json::Value::Number(n) => serde_json::Value::String(n.to_string()),
175                    other => serde_json::Value::String(other.to_string()),
176                };
177                (k.clone(), v)
178            })
179            .collect(),
180    )
181}
182
183/// Every key a planned page touches — upserts and deletes — as canonical
184/// tuples, deduplicated, in page order. What a sink journals before it
185/// applies the plan.
186pub fn plan_keys(
187    plan: &crate::write_mode::WritePlan,
188    key: &[String],
189) -> Vec<crate::write_mode::KeyTuple> {
190    let mut seen = std::collections::HashSet::new();
191    let mut out = Vec::with_capacity(plan.upserts.len() + plan.deletes.len());
192    let upserts = plan
193        .upserts
194        .iter()
195        .filter_map(|r| crate::write_mode::record_key(r, key));
196    for t in upserts.chain(plan.deletes.iter().cloned()) {
197        let c = canonical_key(&t);
198        if seen.insert(key_json(&c)) {
199            out.push(c);
200        }
201    }
202    out
203}
204
205/// Dialect-neutral SQL for the before-image journal and the run-scoped delete.
206///
207/// Each builder takes the dialect's identifier quoting and placeholder style
208/// as function pointers (plain `fn`, so a sink can hold one across an await
209/// in a `Send` future), so the SQL sinks share one tested generator instead of three
210/// hand-written copies. `before_type` is the dialect's JSON-or-text column type
211/// (`JSONB`, `TEXT`, `JSON`); `insert_ignore` is its "skip an existing key"
212/// spelling (`INSERT … ON CONFLICT DO NOTHING`, `INSERT OR IGNORE`,
213/// `INSERT IGNORE`).
214#[derive(Clone, Copy)]
215pub struct JournalSql {
216    pub quote: fn(&str) -> String,
217    pub placeholder: fn(usize) -> String,
218    pub before_type: &'static str,
219    pub insert_prefix: &'static str,
220    pub insert_suffix: &'static str,
221    pub now: &'static str,
222    /// The `key_json` column definition (plus any helper column). A dialect
223    /// with a byte-limited index (MySQL: 3072 bytes) keys on a stored hash of
224    /// the JSON instead of the text itself.
225    pub key_column: &'static str,
226    /// The primary-key clause.
227    pub primary_key: &'static str,
228}
229
230/// The `key_json` definition for dialects that can index the text directly.
231pub const KEY_COLUMN_TEXT: &str = "key_json TEXT NOT NULL";
232/// The primary key over the JSON text itself.
233pub const PRIMARY_KEY_TEXT: &str = "PRIMARY KEY (run_id, table_name, key_json)";
234
235impl JournalSql {
236    fn t(&self) -> String {
237        (self.quote)(RUN_JOURNAL_TABLE)
238    }
239
240    /// `CREATE TABLE IF NOT EXISTS _faucet_run_journal (…)`.
241    pub fn create(&self) -> String {
242        format!(
243            "CREATE TABLE IF NOT EXISTS {t} (run_id VARCHAR(64) NOT NULL, table_name VARCHAR(255) NOT NULL, \
244             {key}, before_json {bt}, recorded_at TIMESTAMP DEFAULT {now}, {pk})",
245            t = self.t(),
246            key = self.key_column,
247            bt = self.before_type,
248            now = self.now,
249            pk = self.primary_key,
250        )
251    }
252
253    /// Insert `rows` journal rows of `(run_id, table_name, key_json,
254    /// before_json)`, skipping keys already journaled for the run — the first
255    /// page to touch a key holds its true pre-run image.
256    pub fn insert(&self, rows: usize) -> String {
257        let mut n = 0usize;
258        let tuples: Vec<String> = (0..rows)
259            .map(|_| {
260                let ph: Vec<String> = (0..4)
261                    .map(|_| {
262                        n += 1;
263                        (self.placeholder)(n)
264                    })
265                    .collect();
266                format!("({})", ph.join(", "))
267            })
268            .collect();
269        format!(
270            "{prefix} INTO {t} (run_id, table_name, key_json, before_json) VALUES {v}{suffix}",
271            prefix = self.insert_prefix,
272            t = self.t(),
273            v = tuples.join(", "),
274            suffix = self.insert_suffix,
275        )
276    }
277
278    /// Every journal row of one run against one table.
279    pub fn select(&self) -> String {
280        format!(
281            "SELECT key_json, before_json FROM {t} WHERE run_id = {p1} AND table_name = {p2}",
282            t = self.t(),
283            p1 = (self.placeholder)(1),
284            p2 = (self.placeholder)(2),
285        )
286    }
287
288    /// Drop one run's journal rows for one table (after a restore).
289    pub fn delete_table(&self) -> String {
290        format!(
291            "DELETE FROM {t} WHERE run_id = {p1} AND table_name = {p2}",
292            t = self.t(),
293            p1 = (self.placeholder)(1),
294            p2 = (self.placeholder)(2),
295        )
296    }
297
298    /// `(k1, k2) IN ((p, p), (p, p), …)` over `tuples` key tuples, placeholders
299    /// numbered from `start + 1`. Returns the predicate and the next free
300    /// placeholder index.
301    pub fn keys_in(&self, key: &[String], tuples: usize, start: usize) -> (String, usize) {
302        let cols: Vec<String> = key.iter().map(|k| (self.quote)(k)).collect();
303        let mut n = start;
304        let groups: Vec<String> = (0..tuples)
305            .map(|_| {
306                let ph: Vec<String> = key
307                    .iter()
308                    .map(|_| {
309                        n += 1;
310                        (self.placeholder)(n)
311                    })
312                    .collect();
313                format!("({})", ph.join(", "))
314            })
315            .collect();
316        (
317            format!("({}) IN ({})", cols.join(", "), groups.join(", ")),
318            n,
319        )
320    }
321
322    /// `DELETE FROM <table> WHERE <run_id_col> = p1` — the append undo.
323    pub fn delete_by_run(&self, table_ref: &str, run_id_col: &str) -> String {
324        format!(
325            "DELETE FROM {table_ref} WHERE {c} = {p}",
326            c = (self.quote)(run_id_col),
327            p = (self.placeholder)(1)
328        )
329    }
330
331    /// `SELECT count(*) FROM <table> WHERE <run_id_col> = p1`.
332    pub fn count_by_run(&self, table_ref: &str, run_id_col: &str) -> String {
333        format!(
334            "SELECT count(*) FROM {table_ref} WHERE {c} = {p}",
335            c = (self.quote)(run_id_col),
336            p = (self.placeholder)(1)
337        )
338    }
339}
340
341/// One journaled key, decoded: the key object and its before-image (`None`
342/// when the key did not exist before the run).
343#[derive(Debug, Clone, PartialEq)]
344pub struct JournalEntry {
345    pub key: serde_json::Map<String, serde_json::Value>,
346    pub before: Option<serde_json::Value>,
347}
348
349impl JournalEntry {
350    /// Decode a journal row's two text columns.
351    pub fn decode(key_json: &str, before_json: Option<&str>) -> Result<Self, crate::FaucetError> {
352        let key: serde_json::Value = serde_json::from_str(key_json).map_err(|e| {
353            crate::FaucetError::Sink(format!("rollback: malformed journal key {key_json:?}: {e}"))
354        })?;
355        let key = match key {
356            serde_json::Value::Object(m) => m,
357            other => {
358                return Err(crate::FaucetError::Sink(format!(
359                    "rollback: journal key is not an object: {other}"
360                )));
361            }
362        };
363        let before = match before_json {
364            None => None,
365            Some(b) => Some(serde_json::from_str(b).map_err(|e| {
366                crate::FaucetError::Sink(format!("rollback: malformed before-image: {e}"))
367            })?),
368        };
369        Ok(Self { key, before })
370    }
371
372    /// The key as a [`KeyTuple`](crate::write_mode::KeyTuple) in `key` order.
373    pub fn tuple(&self, key: &[String]) -> crate::write_mode::KeyTuple {
374        crate::write_mode::KeyTuple(
375            key.iter()
376                .map(|k| {
377                    (
378                        k.clone(),
379                        self.key.get(k).cloned().unwrap_or(serde_json::Value::Null),
380                    )
381                })
382                .collect(),
383        )
384    }
385}
386
387/// Split journal entries into what a restore does: keys to **delete** (the
388/// run created them) and before-images to **write back** (the run changed or
389/// deleted them). Pure.
390pub fn plan_restore(entries: &[JournalEntry]) -> (Vec<&JournalEntry>, Vec<serde_json::Value>) {
391    let mut deletes = Vec::new();
392    let mut restores = Vec::new();
393    for e in entries {
394        match &e.before {
395            None => deletes.push(e),
396            Some(b) => restores.push(b.clone()),
397        }
398    }
399    (deletes, restores)
400}
401
402#[cfg(test)]
403mod tests {
404    use super::*;
405    use crate::write_mode::WriteMode;
406
407    fn q(s: &str) -> String {
408        format!("\"{s}\"")
409    }
410    fn ph(n: usize) -> String {
411        format!("${n}")
412    }
413    fn sql() -> JournalSql {
414        JournalSql {
415            quote: q,
416            placeholder: ph,
417            before_type: "JSONB",
418            insert_prefix: "INSERT",
419            insert_suffix: " ON CONFLICT DO NOTHING",
420            now: "now()",
421            key_column: KEY_COLUMN_TEXT,
422            primary_key: PRIMARY_KEY_TEXT,
423        }
424    }
425
426    #[test]
427    fn journal_sql_shapes() {
428        let j = sql();
429        assert!(
430            j.create()
431                .starts_with("CREATE TABLE IF NOT EXISTS \"_faucet_run_journal\"")
432        );
433        assert!(j.create().contains("before_json JSONB"));
434        assert!(j.create().contains("key_json TEXT NOT NULL"));
435        assert!(
436            j.create()
437                .ends_with("PRIMARY KEY (run_id, table_name, key_json))"),
438            "{}",
439            j.create()
440        );
441        let ins = j.insert(2);
442        assert!(
443            ins.starts_with("INSERT INTO \"_faucet_run_journal\""),
444            "{ins}"
445        );
446        assert!(ins.contains("($1, $2, $3, $4), ($5, $6, $7, $8)"), "{ins}");
447        assert!(ins.ends_with(" ON CONFLICT DO NOTHING"));
448        assert_eq!(
449            j.select(),
450            "SELECT key_json, before_json FROM \"_faucet_run_journal\" WHERE run_id = $1 AND table_name = $2"
451        );
452        assert!(
453            j.delete_table()
454                .starts_with("DELETE FROM \"_faucet_run_journal\" WHERE run_id = $1")
455        );
456        let (pred, next) = j.keys_in(&["a".into(), "b".into()], 2, 1);
457        assert_eq!(pred, "(\"a\", \"b\") IN (($2, $3), ($4, $5))");
458        assert_eq!(next, 5);
459        assert_eq!(
460            j.delete_by_run("\"t\"", "_faucet_run_id"),
461            "DELETE FROM \"t\" WHERE \"_faucet_run_id\" = $1"
462        );
463        assert_eq!(
464            j.count_by_run("\"t\"", "_faucet_run_id"),
465            "SELECT count(*) FROM \"t\" WHERE \"_faucet_run_id\" = $1"
466        );
467    }
468
469    #[test]
470    fn journal_entries_decode_and_plan() {
471        let created = JournalEntry::decode("{\"id\":1}", None).unwrap();
472        let changed = JournalEntry::decode("{\"id\":2}", Some("{\"id\":2,\"v\":\"old\"}")).unwrap();
473        assert_eq!(
474            created.tuple(&["id".into()]).0,
475            vec![("id".to_string(), serde_json::json!(1))]
476        );
477        assert_eq!(
478            created.tuple(&["id".into(), "missing".into()]).0[1],
479            ("missing".to_string(), serde_json::Value::Null)
480        );
481        let entries = [created.clone(), changed.clone()];
482        let (deletes, restores) = plan_restore(&entries);
483        assert_eq!(deletes, vec![&created]);
484        assert_eq!(restores, vec![serde_json::json!({"id": 2, "v": "old"})]);
485        assert!(JournalEntry::decode("nope", None).is_err());
486        assert!(JournalEntry::decode("[1]", None).is_err());
487        assert!(JournalEntry::decode("{\"id\":1}", Some("{broken")).is_err());
488        let kt = crate::write_mode::KeyTuple(vec![
489            ("b".into(), serde_json::json!(2)),
490            ("a".into(), serde_json::json!("x")),
491        ]);
492        assert_eq!(
493            key_json(&kt),
494            "{\"b\":2,\"a\":\"x\"}",
495            "configured key order, not sorted"
496        );
497    }
498
499    #[test]
500    fn plan_keys_are_canonical_and_deduplicated() {
501        let spec = crate::write_mode::WriteSpec {
502            write_mode: WriteMode::Upsert,
503            key: vec!["id".into()],
504            ..Default::default()
505        };
506        let page = vec![
507            serde_json::json!({"id": 7, "v": 1}),
508            serde_json::json!({"id": "7", "v": 2}),
509            serde_json::json!({"id": true}),
510            serde_json::json!({"id": null}),
511        ];
512        let plan = crate::write_mode::plan_writes(&page, &spec);
513        let keys = plan_keys(&plan, &spec.key);
514        // 7 and "7" journal as one key; the null-key row is a plan failure.
515        assert_eq!(keys.len(), 2);
516        assert_eq!(key_json(&keys[0]), "{\"id\":\"7\"}");
517        assert_eq!(key_json(&keys[1]), "{\"id\":\"true\"}");
518        let nested = canonical_key(&crate::write_mode::KeyTuple(vec![
519            ("k".into(), serde_json::json!([1])),
520            ("n".into(), serde_json::Value::Null),
521        ]));
522        assert_eq!(nested.0[0].1, serde_json::json!("[1]"));
523        assert_eq!(nested.0[1].1, serde_json::Value::Null);
524        // Deletes are journaled too.
525        let del = crate::write_mode::WriteSpec {
526            write_mode: WriteMode::Delete,
527            key: vec!["id".into()],
528            ..Default::default()
529        };
530        let plan = crate::write_mode::plan_writes(&[serde_json::json!({"id": 3})], &del);
531        assert_eq!(key_json(&plan_keys(&plan, &del.key)[0]), "{\"id\":\"3\"}");
532    }
533
534    #[test]
535    fn mode_follows_the_write_mode() {
536        assert_eq!(
537            RollbackMode::for_write_mode(WriteMode::Append),
538            RollbackMode::Append
539        );
540        assert_eq!(
541            RollbackMode::for_write_mode(WriteMode::Upsert),
542            RollbackMode::Upsert
543        );
544        assert_eq!(
545            RollbackMode::for_write_mode(WriteMode::Delete),
546            RollbackMode::Upsert
547        );
548        assert_eq!(
549            RollbackMode::for_write_mode(WriteMode::Overwrite),
550            RollbackMode::Overwrite
551        );
552        assert_eq!(RollbackMode::Upsert.as_str(), "upsert");
553        assert_eq!(
554            serde_json::to_string(&RollbackMode::Overwrite).unwrap(),
555            "\"overwrite\""
556        );
557    }
558
559    #[test]
560    fn outcomes_carry_their_reason() {
561        let n = RollbackOutcome::nothing("no rows");
562        assert!(n.applied && n.deleted == 0);
563        let b = RollbackOutcome::blocked(3);
564        assert!(!b.applied && b.conflicts == 3);
565        assert!(b.note.unwrap().contains("--force"));
566        let spec: RollbackWriteSpec = serde_json::from_value(serde_json::json!({
567            "run_id": "r1", "run_id_column": "_faucet_run_id"
568        }))
569        .unwrap();
570        assert!(!spec.journal && !spec.keep_previous);
571        assert!(
572            serde_json::from_value::<RollbackWriteSpec>(serde_json::json!({
573                "run_id": "r1", "run_id_column": "c", "bogus": 1
574            }))
575            .is_err()
576        );
577    }
578}
faucet_core/rollback.rs

faucet_core/
rollback.rs
