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

faucet_core/
rollback.rs
