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    /// A later run wrote this dataset after the one being undone. A journaled
103    /// key with a before-image whose row is now gone was then deleted by that
104    /// later run, so restoring it would resurrect it: it counts as a conflict.
105    #[serde(default)]
106    pub later_runs: bool,
107}
108
109/// What a sink did (or would do) for one dataset.
110#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
111pub struct RollbackOutcome {
112    /// Rows deleted (append: the run's rows; upsert: keys the run created).
113    pub deleted: u64,
114    /// Rows restored to their before-image (upsert), or rows swapped back
115    /// (overwrite).
116    pub restored: u64,
117    /// Keys a later run changed since; without `force` these block the whole
118    /// dataset.
119    pub conflicts: u64,
120    /// Whether changes were applied. `false` for a dry run, or when conflicts
121    /// blocked the rollback.
122    pub applied: bool,
123    /// A human note (e.g. why nothing was applied).
124    #[serde(default, skip_serializing_if = "Option::is_none")]
125    pub note: Option<String>,
126}
127
128impl RollbackOutcome {
129    /// Nothing to undo: the run left no rows / journal / previous table.
130    pub fn nothing(note: impl Into<String>) -> Self {
131        Self {
132            applied: true,
133            note: Some(note.into()),
134            ..Default::default()
135        }
136    }
137
138    /// Blocked by conflicts (no change made).
139    pub fn blocked(conflicts: u64) -> Self {
140        Self {
141            conflicts,
142            applied: false,
143            note: Some(format!(
144                "{conflicts} key(s) were changed by a later run; pass --force to restore them anyway"
145            )),
146            ..Default::default()
147        }
148    }
149}
150
151/// The canonical JSON text of a key tuple — the journal's row identity.
152/// Columns are emitted in the configured key order, so the same key always
153/// serialises identically.
154pub fn key_json(tuple: &crate::write_mode::KeyTuple) -> String {
155    // Built by hand rather than through a `Map`: the map's key order depends
156    // on the `preserve_order` feature, and this text is a primary key.
157    let parts: Vec<String> = tuple
158        .0
159        .iter()
160        .map(|(k, v)| format!("{}:{}", serde_json::Value::String(k.clone()), v))
161        .collect();
162    format!("{{{}}}", parts.join(","))
163}
164
165/// A key tuple with every scalar rendered as its text form, so a key read
166/// back from the database (`7`) and the same key in a source record (`"7"`)
167/// journal identically. Nulls stay null; the SQL sinks bind the text through
168/// the key column's own type, so the restore addresses the right row.
169pub fn canonical_key(tuple: &crate::write_mode::KeyTuple) -> crate::write_mode::KeyTuple {
170    crate::write_mode::KeyTuple(
171        tuple
172            .0
173            .iter()
174            .map(|(k, v)| {
175                let v = match v {
176                    serde_json::Value::Null => serde_json::Value::Null,
177                    serde_json::Value::String(s) => serde_json::Value::String(s.clone()),
178                    serde_json::Value::Bool(b) => serde_json::Value::String(b.to_string()),
179                    serde_json::Value::Number(n) => serde_json::Value::String(n.to_string()),
180                    other => serde_json::Value::String(crate::util::canonical_json(other)),
181                };
182                (k.clone(), v)
183            })
184            .collect(),
185    )
186}
187
188/// Every key a planned page touches — upserts and deletes — as canonical
189/// tuples, deduplicated, in page order. What a sink journals before it
190/// applies the plan.
191pub fn plan_keys(
192    plan: &crate::write_mode::WritePlan,
193    key: &[String],
194) -> Vec<crate::write_mode::KeyTuple> {
195    let mut seen = std::collections::HashSet::new();
196    let mut out = Vec::with_capacity(plan.upserts.len() + plan.deletes.len());
197    let upserts = plan
198        .upserts
199        .iter()
200        .filter_map(|r| crate::write_mode::record_key(r, key));
201    for t in upserts.chain(plan.deletes.iter().cloned()) {
202        let c = canonical_key(&t);
203        if seen.insert(key_json(&c)) {
204            out.push(c);
205        }
206    }
207    out
208}
209
210/// Dialect-neutral SQL for the before-image journal and the run-scoped delete.
211///
212/// Each builder takes the dialect's identifier quoting and placeholder style
213/// as function pointers (plain `fn`, so a sink can hold one across an await
214/// in a `Send` future), so the SQL sinks share one tested generator instead of three
215/// hand-written copies. `before_type` is the dialect's JSON-or-text column type
216/// (`JSONB`, `TEXT`, `JSON`); `insert_ignore` is its "skip an existing key"
217/// spelling (`INSERT … ON CONFLICT DO NOTHING`, `INSERT OR IGNORE`,
218/// `INSERT IGNORE`).
219#[derive(Clone, Copy)]
220pub struct JournalSql {
221    pub quote: fn(&str) -> String,
222    pub placeholder: fn(usize) -> String,
223    pub before_type: &'static str,
224    pub insert_prefix: &'static str,
225    pub insert_suffix: &'static str,
226    pub now: &'static str,
227    /// The `key_json` column definition (plus any helper column). A dialect
228    /// with a byte-limited index (MySQL: 3072 bytes) keys on a stored hash of
229    /// the JSON instead of the text itself.
230    pub key_column: &'static str,
231    /// The primary-key clause.
232    pub primary_key: &'static str,
233}
234
235/// The `key_json` definition for dialects that can index the text directly.
236pub const KEY_COLUMN_TEXT: &str = "key_json TEXT NOT NULL";
237/// The primary key over the JSON text itself.
238pub const PRIMARY_KEY_TEXT: &str = "PRIMARY KEY (run_id, table_name, key_json)";
239
240impl JournalSql {
241    fn t(&self) -> String {
242        (self.quote)(RUN_JOURNAL_TABLE)
243    }
244
245    /// `CREATE TABLE IF NOT EXISTS _faucet_run_journal (…)`.
246    pub fn create(&self) -> String {
247        format!(
248            "CREATE TABLE IF NOT EXISTS {t} (run_id VARCHAR(64) NOT NULL, table_name VARCHAR(255) NOT NULL, \
249             {key}, before_json {bt}, recorded_at TIMESTAMP DEFAULT {now}, {pk})",
250            t = self.t(),
251            key = self.key_column,
252            bt = self.before_type,
253            now = self.now,
254            pk = self.primary_key,
255        )
256    }
257
258    /// Insert `rows` journal rows of `(run_id, table_name, key_json,
259    /// before_json)`, skipping keys already journaled for the run — the first
260    /// page to touch a key holds its true pre-run image.
261    pub fn insert(&self, rows: usize) -> String {
262        let mut n = 0usize;
263        let tuples: Vec<String> = (0..rows)
264            .map(|_| {
265                let ph: Vec<String> = (0..4)
266                    .map(|_| {
267                        n += 1;
268                        (self.placeholder)(n)
269                    })
270                    .collect();
271                format!("({})", ph.join(", "))
272            })
273            .collect();
274        format!(
275            "{prefix} INTO {t} (run_id, table_name, key_json, before_json) VALUES {v}{suffix}",
276            prefix = self.insert_prefix,
277            t = self.t(),
278            v = tuples.join(", "),
279            suffix = self.insert_suffix,
280        )
281    }
282
283    /// Every journal row of one run against one table.
284    pub fn select(&self) -> String {
285        format!(
286            "SELECT key_json, before_json FROM {t} WHERE run_id = {p1} AND table_name = {p2}",
287            t = self.t(),
288            p1 = (self.placeholder)(1),
289            p2 = (self.placeholder)(2),
290        )
291    }
292
293    /// Drop one run's journal rows for one table (after a restore).
294    pub fn delete_table(&self) -> String {
295        format!(
296            "DELETE FROM {t} WHERE run_id = {p1} AND table_name = {p2}",
297            t = self.t(),
298            p1 = (self.placeholder)(1),
299            p2 = (self.placeholder)(2),
300        )
301    }
302
303    /// `(k1, k2) IN ((p, p), (p, p), …)` over `tuples` key tuples, placeholders
304    /// numbered from `start + 1`. Returns the predicate and the next free
305    /// placeholder index.
306    pub fn keys_in(&self, key: &[String], tuples: usize, start: usize) -> (String, usize) {
307        let cols: Vec<String> = key.iter().map(|k| (self.quote)(k)).collect();
308        let mut n = start;
309        let groups: Vec<String> = (0..tuples)
310            .map(|_| {
311                let ph: Vec<String> = key
312                    .iter()
313                    .map(|_| {
314                        n += 1;
315                        (self.placeholder)(n)
316                    })
317                    .collect();
318                format!("({})", ph.join(", "))
319            })
320            .collect();
321        (
322            format!("({}) IN ({})", cols.join(", "), groups.join(", ")),
323            n,
324        )
325    }
326
327    /// `DELETE FROM <table> WHERE <run_id_col> = p1` — the append undo.
328    pub fn delete_by_run(&self, table_ref: &str, run_id_col: &str) -> String {
329        format!(
330            "DELETE FROM {table_ref} WHERE {c} = {p}",
331            c = (self.quote)(run_id_col),
332            p = (self.placeholder)(1)
333        )
334    }
335
336    /// `SELECT count(*) FROM <table> WHERE <run_id_col> = p1`.
337    pub fn count_by_run(&self, table_ref: &str, run_id_col: &str) -> String {
338        format!(
339            "SELECT count(*) FROM {table_ref} WHERE {c} = {p}",
340            c = (self.quote)(run_id_col),
341            p = (self.placeholder)(1)
342        )
343    }
344}
345
346/// One journaled key, decoded: the key object and its before-image (`None`
347/// when the key did not exist before the run).
348#[derive(Debug, Clone, PartialEq)]
349pub struct JournalEntry {
350    pub key: serde_json::Map<String, serde_json::Value>,
351    pub before: Option<serde_json::Value>,
352}
353
354impl JournalEntry {
355    /// Decode a journal row's two text columns.
356    pub fn decode(key_json: &str, before_json: Option<&str>) -> Result<Self, crate::FaucetError> {
357        let key: serde_json::Value = serde_json::from_str(key_json).map_err(|e| {
358            crate::FaucetError::Sink(format!("rollback: malformed journal key {key_json:?}: {e}"))
359        })?;
360        let key = match key {
361            serde_json::Value::Object(m) => m,
362            other => {
363                return Err(crate::FaucetError::Sink(format!(
364                    "rollback: journal key is not an object: {other}"
365                )));
366            }
367        };
368        let before = match before_json {
369            None => None,
370            Some(b) => Some(serde_json::from_str(b).map_err(|e| {
371                crate::FaucetError::Sink(format!("rollback: malformed before-image: {e}"))
372            })?),
373        };
374        Ok(Self { key, before })
375    }
376
377    /// The key as a [`KeyTuple`](crate::write_mode::KeyTuple) in `key` order.
378    pub fn tuple(&self, key: &[String]) -> crate::write_mode::KeyTuple {
379        crate::write_mode::KeyTuple(
380            key.iter()
381                .map(|k| {
382                    (
383                        k.clone(),
384                        self.key.get(k).cloned().unwrap_or(serde_json::Value::Null),
385                    )
386                })
387                .collect(),
388        )
389    }
390}
391
392/// Split journal entries into what a restore does: keys to **delete** (the
393/// run created them) and before-images to **write back** (the run changed or
394/// deleted them). Pure.
395/// Journaled keys whose undo would re-insert a before-image. When a later run
396/// wrote the dataset, each of these that no longer has a row is a conflict
397/// (see [`RollbackOptions::later_runs`]).
398pub fn restorable(entries: &[JournalEntry]) -> Vec<&JournalEntry> {
399    entries.iter().filter(|e| e.before.is_some()).collect()
400}
401
402pub fn plan_restore(entries: &[JournalEntry]) -> (Vec<&JournalEntry>, Vec<serde_json::Value>) {
403    let mut deletes = Vec::new();
404    let mut restores = Vec::new();
405    for e in entries {
406        match &e.before {
407            None => deletes.push(e),
408            Some(b) => restores.push(b.clone()),
409        }
410    }
411    (deletes, restores)
412}
413
414#[cfg(test)]
415mod tests {
416    use super::*;
417    use crate::write_mode::WriteMode;
418
419    #[test]
420    fn restorable_keeps_only_keys_with_a_before_image() {
421        let entries = vec![
422            JournalEntry::decode(r#"{"id":"1"}"#, None).unwrap(),
423            JournalEntry::decode(r#"{"id":"2"}"#, Some(r#"{"id":2,"v":"a"}"#)).unwrap(),
424        ];
425        let kept = restorable(&entries);
426        assert_eq!(kept.len(), 1);
427        assert_eq!(kept[0].key["id"], "2");
428    }
429
430    #[test]
431    fn later_runs_defaults_off_when_absent_from_the_wire() {
432        let opts: RollbackOptions = serde_json::from_value(serde_json::json!({
433            "run_id_column": "_faucet_run_id",
434            "mode": "upsert",
435            "force": false,
436            "dry_run": false
437        }))
438        .unwrap();
439        assert!(!opts.later_runs);
440    }
441
442    fn q(s: &str) -> String {
443        format!("\"{s}\"")
444    }
445    fn ph(n: usize) -> String {
446        format!("${n}")
447    }
448    fn sql() -> JournalSql {
449        JournalSql {
450            quote: q,
451            placeholder: ph,
452            before_type: "JSONB",
453            insert_prefix: "INSERT",
454            insert_suffix: " ON CONFLICT DO NOTHING",
455            now: "now()",
456            key_column: KEY_COLUMN_TEXT,
457            primary_key: PRIMARY_KEY_TEXT,
458        }
459    }
460
461    #[test]
462    fn journal_sql_shapes() {
463        let j = sql();
464        assert!(
465            j.create()
466                .starts_with("CREATE TABLE IF NOT EXISTS \"_faucet_run_journal\"")
467        );
468        assert!(j.create().contains("before_json JSONB"));
469        assert!(j.create().contains("key_json TEXT NOT NULL"));
470        assert!(
471            j.create()
472                .ends_with("PRIMARY KEY (run_id, table_name, key_json))"),
473            "{}",
474            j.create()
475        );
476        let ins = j.insert(2);
477        assert!(
478            ins.starts_with("INSERT INTO \"_faucet_run_journal\""),
479            "{ins}"
480        );
481        assert!(ins.contains("($1, $2, $3, $4), ($5, $6, $7, $8)"), "{ins}");
482        assert!(ins.ends_with(" ON CONFLICT DO NOTHING"));
483        assert_eq!(
484            j.select(),
485            "SELECT key_json, before_json FROM \"_faucet_run_journal\" WHERE run_id = $1 AND table_name = $2"
486        );
487        assert!(
488            j.delete_table()
489                .starts_with("DELETE FROM \"_faucet_run_journal\" WHERE run_id = $1")
490        );
491        let (pred, next) = j.keys_in(&["a".into(), "b".into()], 2, 1);
492        assert_eq!(pred, "(\"a\", \"b\") IN (($2, $3), ($4, $5))");
493        assert_eq!(next, 5);
494        assert_eq!(
495            j.delete_by_run("\"t\"", "_faucet_run_id"),
496            "DELETE FROM \"t\" WHERE \"_faucet_run_id\" = $1"
497        );
498        assert_eq!(
499            j.count_by_run("\"t\"", "_faucet_run_id"),
500            "SELECT count(*) FROM \"t\" WHERE \"_faucet_run_id\" = $1"
501        );
502    }
503
504    #[test]
505    fn journal_entries_decode_and_plan() {
506        let created = JournalEntry::decode("{\"id\":1}", None).unwrap();
507        let changed = JournalEntry::decode("{\"id\":2}", Some("{\"id\":2,\"v\":\"old\"}")).unwrap();
508        assert_eq!(
509            created.tuple(&["id".into()]).0,
510            vec![("id".to_string(), serde_json::json!(1))]
511        );
512        assert_eq!(
513            created.tuple(&["id".into(), "missing".into()]).0[1],
514            ("missing".to_string(), serde_json::Value::Null)
515        );
516        let entries = [created.clone(), changed.clone()];
517        let (deletes, restores) = plan_restore(&entries);
518        assert_eq!(deletes, vec![&created]);
519        assert_eq!(restores, vec![serde_json::json!({"id": 2, "v": "old"})]);
520        assert!(JournalEntry::decode("nope", None).is_err());
521        assert!(JournalEntry::decode("[1]", None).is_err());
522        assert!(JournalEntry::decode("{\"id\":1}", Some("{broken")).is_err());
523        let kt = crate::write_mode::KeyTuple(vec![
524            ("b".into(), serde_json::json!(2)),
525            ("a".into(), serde_json::json!("x")),
526        ]);
527        assert_eq!(
528            key_json(&kt),
529            "{\"b\":2,\"a\":\"x\"}",
530            "configured key order, not sorted"
531        );
532    }
533
534    #[test]
535    fn plan_keys_are_canonical_and_deduplicated() {
536        let spec = crate::write_mode::WriteSpec {
537            write_mode: WriteMode::Upsert,
538            key: vec!["id".into()],
539            ..Default::default()
540        };
541        let page = vec![
542            serde_json::json!({"id": 7, "v": 1}),
543            serde_json::json!({"id": "7", "v": 2}),
544            serde_json::json!({"id": true}),
545            serde_json::json!({"id": null}),
546        ];
547        let plan = crate::write_mode::plan_writes(&page, &spec);
548        let keys = plan_keys(&plan, &spec.key);
549        // 7 and "7" journal as one key; the null-key row is a plan failure.
550        assert_eq!(keys.len(), 2);
551        assert_eq!(key_json(&keys[0]), "{\"id\":\"7\"}");
552        assert_eq!(key_json(&keys[1]), "{\"id\":\"true\"}");
553        let nested = canonical_key(&crate::write_mode::KeyTuple(vec![
554            ("k".into(), serde_json::json!([1])),
555            ("n".into(), serde_json::Value::Null),
556        ]));
557        assert_eq!(nested.0[0].1, serde_json::json!("[1]"));
558        assert_eq!(nested.0[1].1, serde_json::Value::Null);
559        // Deletes are journaled too.
560        let del = crate::write_mode::WriteSpec {
561            write_mode: WriteMode::Delete,
562            key: vec!["id".into()],
563            ..Default::default()
564        };
565        let plan = crate::write_mode::plan_writes(&[serde_json::json!({"id": 3})], &del);
566        assert_eq!(key_json(&plan_keys(&plan, &del.key)[0]), "{\"id\":\"3\"}");
567    }
568
569    #[test]
570    fn mode_follows_the_write_mode() {
571        assert_eq!(
572            RollbackMode::for_write_mode(WriteMode::Append),
573            RollbackMode::Append
574        );
575        assert_eq!(
576            RollbackMode::for_write_mode(WriteMode::Upsert),
577            RollbackMode::Upsert
578        );
579        assert_eq!(
580            RollbackMode::for_write_mode(WriteMode::Delete),
581            RollbackMode::Upsert
582        );
583        assert_eq!(
584            RollbackMode::for_write_mode(WriteMode::Overwrite),
585            RollbackMode::Overwrite
586        );
587        assert_eq!(RollbackMode::Upsert.as_str(), "upsert");
588        assert_eq!(
589            serde_json::to_string(&RollbackMode::Overwrite).unwrap(),
590            "\"overwrite\""
591        );
592    }
593
594    #[test]
595    fn outcomes_carry_their_reason() {
596        let n = RollbackOutcome::nothing("no rows");
597        assert!(n.applied && n.deleted == 0);
598        let b = RollbackOutcome::blocked(3);
599        assert!(!b.applied && b.conflicts == 3);
600        assert!(b.note.unwrap().contains("--force"));
601        let spec: RollbackWriteSpec = serde_json::from_value(serde_json::json!({
602            "run_id": "r1", "run_id_column": "_faucet_run_id"
603        }))
604        .unwrap();
605        assert!(!spec.journal && !spec.keep_previous);
606        assert!(
607            serde_json::from_value::<RollbackWriteSpec>(serde_json::json!({
608                "run_id": "r1", "run_id_column": "c", "bogus": 1
609            }))
610            .is_err()
611        );
612    }
613}
faucet_core/rollback.rs

faucet_core/
rollback.rs
