ignition_core/deploy/apply/
journal.rs

1//! The on-box apply journal, format v1 (Phase 23, D-04, D-07, D-09).
2//!
3//! Written on the gateway host before the first file moves and removed when
4//! the run is finalized; a journal that is still there means an apply was
5//! interrupted. It is read by a later runner (`rollback --run`, recovery, the
6//! `deploy plan` refusal), so the layout is one-way and strict, in the
7//! manner of the declared-names ledger ([`crate::deploy::ledger::Ledger`]):
8//! at most [`JOURNAL_MAX_BYTES`] (8 MiB), strict UTF-8, no duplicate key,
9//! bounded depth, no unknown field, every field present.
10//!
11//! ```json
12//! {
13//!   "schema_version": 1,
14//!   "run_id": "<[A-Za-z0-9._-]{1,128}>",
15//!   "gateway": "<target gateway>",
16//!   "mode": "<target mode>",
17//!   "phase": "snapshotting | ready | writing | written | restoring",
18//!   "pre_fingerprint": "<64 lowercase hex>",
19//!   "touched": ["<collection path or ledger.json>", "..."]
20//! }
21//! ```
22//!
23//! `touched` is every path apply writes or deletes plus the ledger (key
24//! `ledger.json`), byte-sorted and unique; each path is a collection path
25//! (`mirror_path_problem`) or exactly `ledger.json`. A refusal is
26//! `deploy_run_invalid` naming the location and the problem, never a file
27//! value beyond a listed path, and never a fallback to a default (T-23-11).
28
29use std::collections::BTreeMap;
30
31use serde::de::DeserializeOwned;
32use serde::{Deserialize, Serialize};
33
34use super::fingerprint::{LEDGER_KEY, fingerprint};
35use super::run_store::RunId;
36use crate::deploy::layout::RelPath;
37use crate::deploy::ledger::LedgerRoot;
38use crate::deploy::node::{self, Node};
39use crate::deploy::target::NodeTarget;
40use crate::deploy::transport::{TreeReader, mirror_path_problem};
41use crate::error::CoreError;
42
43pub use crate::deploy::ledger::journal_location;
44
45/// File name of the journal.
46pub const JOURNAL_FILE: &str = "journal.json";
47
48/// The largest journal accepted (8 MiB), checked before any parse.
49pub const JOURNAL_MAX_BYTES: usize = 8_388_608;
50
51/// How far an interrupted apply got.
52#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
53#[serde(rename_all = "snake_case")]
54pub enum JournalPhase {
55    /// Pre-images are being copied; no live file has moved.
56    Snapshotting,
57    /// Every pre-image is copied and the staging is complete; no live file
58    /// has moved.
59    Ready,
60    /// Live files are being moved.
61    Writing,
62    /// Every move is done; the run is not finalized.
63    Written,
64    /// A restore is under way.
65    Restoring,
66}
67
68impl JournalPhase {
69    /// Whether no live file can have moved yet (D-10): `snapshotting` and
70    /// `ready`.
71    pub fn before_first_move(self) -> bool {
72        matches!(self, JournalPhase::Snapshotting | JournalPhase::Ready)
73    }
74
75    /// The phase's name as the journal file spells it.
76    pub fn as_str(self) -> &'static str {
77        match self {
78            JournalPhase::Snapshotting => "snapshotting",
79            JournalPhase::Ready => "ready",
80            JournalPhase::Writing => "writing",
81            JournalPhase::Written => "written",
82            JournalPhase::Restoring => "restoring",
83        }
84    }
85}
86
87/// The only schema version this build reads and writes.
88const SCHEMA_VERSION: u32 = 1;
89
90/// The journal's top-level keys, in write order.
91const FIELDS: [&str; 7] = [
92    "schema_version",
93    "run_id",
94    "gateway",
95    "mode",
96    "phase",
97    "pre_fingerprint",
98    "touched",
99];
100
101/// The journal.
102#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
103#[serde(deny_unknown_fields)]
104pub struct Journal {
105    /// Always 1.
106    pub schema_version: u32,
107    /// The run this journal belongs to.
108    pub run_id: String,
109    /// The gateway directory name (the target's `gateway`).
110    pub gateway: String,
111    /// The deployment mode (the target's `mode`).
112    pub mode: String,
113    /// How far the run got.
114    pub phase: JournalPhase,
115    /// The touched-set fingerprint before the run, 64 lowercase hex digits.
116    pub pre_fingerprint: String,
117    /// The touched paths, byte-sorted and unique.
118    pub touched: Vec<String>,
119}
120
121impl Journal {
122    /// Parse and validate `bytes` read from `location`, for the target
123    /// `gateway` and `mode`.
124    pub fn parse(
125        location: &str,
126        bytes: &[u8],
127        gateway: &str,
128        mode: &str,
129    ) -> Result<Journal, CoreError> {
130        let bad = |problem: String| refuse(location, bytes, problem);
131        let journal: Journal = parse_strict(
132            location,
133            bytes,
134            JOURNAL_MAX_BYTES,
135            &FIELDS,
136            &[(
137                "phase",
138                &["snapshotting", "ready", "writing", "written", "restoring"],
139            )],
140        )?
141        .0;
142        check_header(
143            journal.schema_version,
144            &journal.run_id,
145            (&journal.gateway, gateway),
146            (&journal.mode, mode),
147        )
148        .map_err(bad)?;
149        if !is_hex64(&journal.pre_fingerprint) {
150            return Err(bad(
151                "pre_fingerprint is not 64 lowercase hex digits".to_string()
152            ));
153        }
154        check_touched(journal.touched.iter().map(String::as_str)).map_err(bad)?;
155        Ok(journal)
156    }
157
158    /// The canonical bytes: pretty JSON in field order plus one LF.
159    pub fn to_bytes(&self) -> Vec<u8> {
160        canonical_bytes(self)
161    }
162}
163
164/// What recovering an interrupted run does (D-10).
165#[derive(Debug, Clone, Copy, PartialEq, Eq)]
166pub enum RecoveryAction {
167    /// No move began and the touched set is what it was before the run:
168    /// remove the journal, the partial snapshot and the staging, and restore
169    /// nothing.
170    ClearWithoutRestore,
171    /// Put the snapshot back, scan, then remove the journal.
172    Restore,
173}
174
175impl RecoveryAction {
176    /// The action's name in the recovery run's events.
177    pub fn as_str(self) -> &'static str {
178        match self {
179            RecoveryAction::ClearWithoutRestore => "clear",
180            RecoveryAction::Restore => "restore",
181        }
182    }
183}
184
185/// Decide the recovery of `journal` from `live`, the live hash of each path of
186/// its touched list (`None` when absent). A journal whose phase is before the
187/// first move and whose live fingerprint equals its pre-run fingerprint died
188/// before it changed anything; every other journal is restored (D-10). There
189/// is no third answer: nothing discards a journal that might sit over a
190/// partial tree (D-12).
191pub fn recovery_action(
192    journal: &Journal,
193    live: &BTreeMap<String, Option<String>>,
194) -> RecoveryAction {
195    if journal.phase.before_first_move() && fingerprint(live) == journal.pre_fingerprint {
196        RecoveryAction::ClearWithoutRestore
197    } else {
198        RecoveryAction::Restore
199    }
200}
201
202/// The journal on the box of `gateway` and `mode`, read through `reader` and
203/// parsed strictly (DAPL-09, D-09). `None` means no interrupted run. A journal
204/// that cannot be read or parsed is `deploy_run_invalid` naming the location:
205/// plan, apply and rollback never go on as if it were absent.
206pub async fn journal_on_box(
207    reader: &dyn TreeReader,
208    node: &NodeTarget,
209    gateway: &str,
210    mode: &str,
211) -> Result<Option<Journal>, CoreError> {
212    let Some(bytes) = reader.read_journal(gateway, mode).await? else {
213        return Ok(None);
214    };
215    let location = journal_location(LedgerRoot::for_node(node), gateway, mode);
216    Journal::parse(&location, &bytes, gateway, mode).map(Some)
217}
218
219/// The refusal for `journal`: `deploy_journal_present` naming its run, the
220/// location and (through its message) `ign deploy rollback --run <id> --env
221/// <env>`.
222pub fn journal_present(
223    journal: &Journal,
224    env: &str,
225    node: &NodeTarget,
226    gateway: &str,
227    mode: &str,
228) -> CoreError {
229    CoreError::DeployJournalPresent {
230        run_id: journal.run_id.clone(),
231        env: env.to_string(),
232        location: journal_location(LedgerRoot::for_node(node), gateway, mode),
233    }
234}
235
236/// The canonical bytes of `value`: pretty JSON plus one LF.
237pub(super) fn canonical_bytes(value: &impl Serialize) -> Vec<u8> {
238    let mut bytes = serde_json::to_vec_pretty(value).unwrap_or_default();
239    bytes.push(b'\n');
240    bytes
241}
242
243/// Whether `text` is 64 lowercase hex digits.
244pub(super) fn is_hex64(text: &str) -> bool {
245    text.len() == 64
246        && text
247            .bytes()
248            .all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b))
249}
250
251/// At most 64 printable characters of `text`, for a problem message.
252pub(super) fn shown(text: &str) -> String {
253    text.chars().filter(|c| !c.is_control()).take(64).collect()
254}
255
256/// The refusal for a journal or record at `location`: `deploy_run_invalid`
257/// naming the location and the problem. The run id is the file's own when it
258/// is a valid id (the rule limits it to a safe charset), else `unknown`.
259pub(super) fn refuse(location: &str, bytes: &[u8], problem: String) -> CoreError {
260    let run_id = serde_json::from_slice::<serde_json::Value>(bytes)
261        .ok()
262        .and_then(|v| v.get("run_id")?.as_str().map(str::to_string))
263        .filter(|id| RunId::parse(id).is_ok())
264        .unwrap_or_else(|| "unknown".to_string());
265    CoreError::DeployRunInvalid {
266        run_id,
267        problem: format!("{location}: {problem}"),
268    }
269}
270
271/// The checks every format applies before it deserializes: not empty, within
272/// `cap`, strict UTF-8 without a repeated key or excess depth, a JSON object,
273/// no unknown key, every key of `fields` present, and each of `enums` one of
274/// its allowed strings. Then the typed read. Returns the value and the
275/// order-preserving document for per-element checks.
276pub(super) fn parse_strict<T: DeserializeOwned>(
277    location: &str,
278    bytes: &[u8],
279    cap: usize,
280    fields: &[&str],
281    enums: &[(&str, &[&str])],
282) -> Result<(T, Node), CoreError> {
283    let bad = |problem: String| refuse(location, bytes, problem);
284    if bytes.is_empty() {
285        return Err(bad("the file is empty".to_string()));
286    }
287    if bytes.len() > cap {
288        return Err(bad(format!(
289            "the file is {} bytes, over the {cap} byte cap",
290            bytes.len()
291        )));
292    }
293    let (_, root) = node::classify(bytes).map_err(bad)?;
294    let Node::Obj(members) = &root else {
295        return Err(bad("the top level is not a JSON object".to_string()));
296    };
297    for (key, _) in members {
298        if !fields.contains(&key.as_str()) {
299            return Err(bad(format!("unknown field `{}`", shown(key))));
300        }
301    }
302    for field in fields {
303        if root.get(field).is_none() {
304            return Err(bad(format!("missing field `{field}`")));
305        }
306    }
307    for (field, allowed) in enums {
308        let ok = root
309            .get(field)
310            .and_then(Node::as_str)
311            .is_some_and(|v| allowed.contains(&v));
312        if !ok {
313            return Err(bad(format!(
314                "`{field}` is not one of {}",
315                allowed.join(", ")
316            )));
317        }
318    }
319    let value: T = serde_json::from_slice(bytes).map_err(|e| {
320        bad(format!(
321            "a field has the wrong type (line {} column {})",
322            e.line(),
323            e.column()
324        ))
325    })?;
326    Ok((value, root))
327}
328
329/// The header checks both formats share: the schema version, the run id and
330/// that the file names this target's gateway and mode. The problem text never
331/// echoes the foreign names.
332pub(super) fn check_header(
333    schema_version: u32,
334    run_id: &str,
335    gateway: (&str, &str),
336    mode: (&str, &str),
337) -> Result<(), String> {
338    if schema_version != SCHEMA_VERSION {
339        return Err(format!(
340            "schema_version is not {SCHEMA_VERSION}, the only version this build reads"
341        ));
342    }
343    if RunId::parse(run_id).is_err() {
344        return Err(
345            "run_id is not 1 to 128 characters of [A-Za-z0-9._-] without a leading dot".to_string(),
346        );
347    }
348    if gateway.0 != gateway.1 {
349        return Err("gateway names a different gateway than the target".to_string());
350    }
351    if mode.0 != mode.1 {
352        return Err("mode names a different mode than the target".to_string());
353    }
354    Ok(())
355}
356
357/// The `touched` rules: each path is a collection path or exactly
358/// `ledger.json`, and the list is byte-sorted and unique.
359pub(super) fn check_touched<'a>(paths: impl Iterator<Item = &'a str>) -> Result<(), String> {
360    let mut previous: Option<&str> = None;
361    for (index, path) in paths.enumerate() {
362        let listed = shown(path);
363        if path != LEDGER_KEY {
364            let collection = RelPath::new(path)
365                .ok()
366                .is_some_and(|rel| mirror_path_problem(&rel).is_none());
367            if !collection {
368                return Err(format!(
369                    "touched[{index}] `{listed}` is neither a collection path nor `{LEDGER_KEY}`"
370                ));
371            }
372        }
373        if let Some(prev) = previous {
374            if path == prev {
375                return Err(format!(
376                    "touched[{index}] `{listed}` is listed twice (duplicate)"
377                ));
378            }
379            if path.as_bytes() < prev.as_bytes() {
380                return Err(format!(
381                    "touched is not byte-sorted: `{listed}` follows `{}`",
382                    shown(prev)
383                ));
384            }
385        }
386        previous = Some(path);
387    }
388    Ok(())
389}
390
391#[cfg(test)]
392mod tests {
393    use super::*;
394    use serde_json::{Value, json};
395
396    const GW: &str = "tracer-gw";
397    const MODE: &str = "dev";
398    const LOC: &str = "~/ign-deploy/tracer-gw/dev/journal.json";
399    const HASH: &str = "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef";
400
401    fn good() -> Journal {
402        Journal {
403            schema_version: 1,
404            run_id: "20261015T142233Z-a1b2c3".to_string(),
405            gateway: GW.to_string(),
406            mode: MODE.to_string(),
407            phase: JournalPhase::Writing,
408            pre_fingerprint: HASH.to_string(),
409            touched: vec![
410                "ledger.json".to_string(),
411                "mod/type/a.json".to_string(),
412                "mod/type/b.json".to_string(),
413            ],
414        }
415    }
416
417    fn with(edit: impl FnOnce(&mut Value)) -> Vec<u8> {
418        let mut value: Value = serde_json::from_slice(&good().to_bytes()).expect("good is JSON");
419        edit(&mut value);
420        serde_json::to_vec(&value).expect("serialize")
421    }
422
423    fn refusal(bytes: &[u8]) -> String {
424        let err = Journal::parse(LOC, bytes, GW, MODE).expect_err("must be refused");
425        assert_eq!(err.code(), "deploy_run_invalid", "{err}");
426        let text = err.to_string();
427        assert!(text.contains(LOC), "the refusal names the location: {text}");
428        text
429    }
430
431    #[test]
432    fn journal_roundtrip() {
433        let journal = good();
434        let bytes = journal.to_bytes();
435        assert!(bytes.ends_with(b"}\n"));
436        let parsed = Journal::parse(LOC, &bytes, GW, MODE).expect("parses");
437        assert_eq!(parsed, journal);
438        assert_eq!(parsed.to_bytes(), bytes, "byte-identical");
439        for phase in [
440            JournalPhase::Snapshotting,
441            JournalPhase::Ready,
442            JournalPhase::Writing,
443            JournalPhase::Written,
444            JournalPhase::Restoring,
445        ] {
446            let j = Journal { phase, ..good() };
447            assert_eq!(
448                Journal::parse(LOC, &j.to_bytes(), GW, MODE).expect("parses"),
449                j
450            );
451        }
452        let empty = Journal {
453            touched: vec![],
454            ..good()
455        };
456        assert_eq!(
457            Journal::parse(LOC, &empty.to_bytes(), GW, MODE).expect("parses"),
458            empty
459        );
460    }
461
462    #[test]
463    fn the_layout_is_pinned() {
464        let text = String::from_utf8(good().to_bytes()).expect("utf-8");
465        let expected = format!(
466            "{{\n  \"schema_version\": 1,\n  \"run_id\": \"20261015T142233Z-a1b2c3\",\n  \"gateway\": \"tracer-gw\",\n  \"mode\": \"dev\",\n  \"phase\": \"writing\",\n  \"pre_fingerprint\": \"{HASH}\",\n  \"touched\": [\n    \"ledger.json\",\n    \"mod/type/a.json\",\n    \"mod/type/b.json\"\n  ]\n}}\n"
467        );
468        assert_eq!(text, expected);
469    }
470
471    #[test]
472    fn phases_before_the_first_move() {
473        assert!(JournalPhase::Snapshotting.before_first_move());
474        assert!(JournalPhase::Ready.before_first_move());
475        assert!(!JournalPhase::Writing.before_first_move());
476        assert!(!JournalPhase::Written.before_first_move());
477        assert!(!JournalPhase::Restoring.before_first_move());
478    }
479
480    fn live(pairs: &[(&str, Option<&str>)]) -> std::collections::BTreeMap<String, Option<String>> {
481        pairs
482            .iter()
483            .map(|(path, hash)| ((*path).to_string(), hash.map(str::to_string)))
484            .collect()
485    }
486
487    #[test]
488    fn recovery_action_clears_only_an_untouched_run() {
489        use super::super::fingerprint::fingerprint;
490        let pre = live(&[("mod/type/a.json", Some(HASH)), ("mod/type/b.json", None)]);
491        let journal = |phase| Journal {
492            phase,
493            pre_fingerprint: fingerprint(&pre),
494            touched: pre.keys().cloned().collect(),
495            ..good()
496        };
497        let other = live(&[
498            ("mod/type/a.json", Some(HASH)),
499            ("mod/type/b.json", Some(HASH)),
500        ]);
501        for (phase, cleared) in [
502            (JournalPhase::Snapshotting, true),
503            (JournalPhase::Ready, true),
504            (JournalPhase::Writing, false),
505            (JournalPhase::Written, false),
506            (JournalPhase::Restoring, false),
507        ] {
508            let j = journal(phase);
509            let same = recovery_action(&j, &pre);
510            let changed = recovery_action(&j, &other);
511            assert_eq!(
512                same,
513                if cleared {
514                    RecoveryAction::ClearWithoutRestore
515                } else {
516                    RecoveryAction::Restore
517                },
518                "{phase:?}, live equals the pre-run state"
519            );
520            assert_eq!(
521                changed,
522                RecoveryAction::Restore,
523                "{phase:?}, live differs from the pre-run state"
524            );
525        }
526    }
527
528    #[test]
529    fn journal_location_follows_the_ledger_root() {
530        use crate::deploy::ledger::LedgerRoot;
531        assert_eq!(
532            journal_location(LedgerRoot::AccountHome, GW, MODE),
533            "~/ign-deploy/tracer-gw/dev/journal.json"
534        );
535        assert_eq!(
536            journal_location(LedgerRoot::DataDir, GW, MODE),
537            "ign-deploy/tracer-gw/dev/journal.json"
538        );
539    }
540
541    #[test]
542    fn refuses_an_empty_file() {
543        assert!(refusal(b"").contains("empty"));
544    }
545
546    #[test]
547    fn refuses_a_file_over_the_cap() {
548        let big = vec![b' '; JOURNAL_MAX_BYTES + 1];
549        assert!(refusal(&big).contains("cap"));
550    }
551
552    #[test]
553    fn refuses_invalid_utf8() {
554        refusal(b"{\"schema_version\": 1, \"run_id\": \"\xff\"}");
555    }
556
557    #[test]
558    fn refuses_a_duplicate_key() {
559        let text = String::from_utf8(good().to_bytes()).expect("utf-8");
560        let dup = text.replacen(
561            "\"mode\": \"dev\"",
562            "\"mode\": \"dev\",\n  \"mode\": \"dev\"",
563            1,
564        );
565        refusal(dup.as_bytes());
566    }
567
568    #[test]
569    fn refuses_an_unknown_field() {
570        let text = refusal(&with(|v| {
571            v["extra"] = json!(1);
572        }));
573        assert!(text.contains("extra"), "{text}");
574    }
575
576    #[test]
577    fn refuses_a_missing_field() {
578        let text = refusal(&with(|v| {
579            v.as_object_mut().expect("object").remove("phase");
580        }));
581        assert!(text.contains("phase"), "{text}");
582    }
583
584    #[test]
585    fn refuses_another_schema_version() {
586        refusal(&with(|v| v["schema_version"] = json!(2)));
587    }
588
589    #[test]
590    fn refuses_another_gateway() {
591        refusal(&with(|v| v["gateway"] = json!("other-gw")));
592    }
593
594    #[test]
595    fn refuses_another_mode() {
596        refusal(&with(|v| v["mode"] = json!("prod")));
597    }
598
599    #[test]
600    fn refuses_a_bad_run_id() {
601        for bad in ["", "../x", ".hidden", "a b"] {
602            refusal(&with(|v| v["run_id"] = json!(bad)));
603        }
604    }
605
606    #[test]
607    fn refuses_an_unknown_phase() {
608        let text = refusal(&with(|v| v["phase"] = json!("halfway")));
609        assert!(text.contains("phase"), "{text}");
610    }
611
612    #[test]
613    fn refuses_unsorted_touched() {
614        let text = refusal(&with(|v| {
615            v["touched"] = json!(["mod/type/b.json", "mod/type/a.json"]);
616        }));
617        assert!(text.contains("sorted"), "{text}");
618    }
619
620    #[test]
621    fn refuses_duplicate_touched() {
622        let text = refusal(&with(|v| {
623            v["touched"] = json!(["mod/type/a.json", "mod/type/a.json"]);
624        }));
625        assert!(text.contains("duplicate"), "{text}");
626    }
627
628    #[test]
629    fn refuses_a_touched_path_that_is_not_a_collection_path() {
630        for bad in [
631            "a/b",
632            "mod/type/.hidden/x",
633            "mod/type/../x",
634            "/abs/x/y",
635            "mod/type/resource.json/../../z",
636            "mod/type/config-mode.json",
637            "ledger.json.bak",
638        ] {
639            refusal(&with(|v| v["touched"] = json!([bad])));
640        }
641    }
642
643    #[test]
644    fn refuses_a_pre_fingerprint_that_is_not_hex() {
645        for bad in [
646            "",
647            "abc",
648            &HASH.to_uppercase(),
649            &format!("{HASH}0"),
650            "g".repeat(64).as_str(),
651        ] {
652            refusal(&with(|v| v["pre_fingerprint"] = json!(bad)));
653        }
654    }
655
656    #[test]
657    fn refuses_a_wrong_typed_field() {
658        refusal(&with(|v| v["touched"] = json!("mod/type/a.json")));
659        refusal(&with(|v| v["schema_version"] = json!("1")));
660    }
661
662    #[test]
663    fn refuses_a_top_level_that_is_not_an_object() {
664        refusal(b"[]");
665    }
666
667    #[test]
668    fn the_refusal_never_echoes_foreign_values() {
669        let text = refusal(&with(|v| v["gateway"] = json!("secret-gateway-name")));
670        assert!(!text.contains("secret-gateway-name"), "{text}");
671    }
672}
ignition_core/deploy/apply/journal.rs

ignition_core/deploy/apply/
journal.rs
