1use std::collections::BTreeMap;
64use std::io::Read as _;
65use std::path::Path;
66
67use serde::Serialize;
68use serde_json::{Value, json};
69
70use super::converge::RestoreSurfaces;
71use super::fingerprint::{LEDGER_KEY, fingerprint};
72use super::hold::{Clock, HeldLock, HoldModel};
73use super::host::ApplyHost;
74use super::journal::{
75 Journal, RecoveryAction, journal_location, journal_on_box, journal_present, recovery_action,
76};
77use super::preflight::{DesignersPolicy, PreflightInput, run_preflights};
78use super::record::{RECORD_FILE, RecordState, RunRecord, TouchedEntry};
79use super::run_store::{EventLog, RunDir, RunId, run_dir_rel};
80use super::{
81 ApplyEvent, ApplyObserver, RestoreContext, RestoreInput, Steps, restart_report,
82 restore_backup_phase, restore_baseline, restore_snapshot,
83};
84use crate::actions::workspace::rfc3339_now_utc;
85use crate::client::scan::ScanKind;
86use crate::client::{DeployApi, GatewayApi};
87use crate::deploy::component_problem;
88use crate::deploy::layout::is_pair_unsynced_tag_path;
89use crate::deploy::ledger::LedgerRoot;
90use crate::deploy::pair::{BackupNode, BackupPolicy, BackupReport};
91use crate::deploy::target::Target;
92use crate::deploy::transport::TreeReader;
93use crate::error::{APPLY_LIST_SHOWN, CoreError, RollbackFailure, RollbackScope};
94use crate::poll::PollConfig;
95
96const LOCAL_FILE_MAX_BYTES: u64 = 8_388_608;
98
99#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
101#[serde(rename_all = "snake_case")]
102pub enum RollbackKind {
103 Restored,
105 Cleared,
109}
110
111impl RollbackKind {
112 pub fn as_str(self) -> &'static str {
114 match self {
115 RollbackKind::Restored => "restored",
116 RollbackKind::Cleared => "cleared",
117 }
118 }
119}
120
121pub struct RollbackContext<'a> {
124 pub env: &'a str,
126 pub repo: &'a Path,
128 pub target_file: &'a str,
130 pub target: &'a Target,
132 pub reader: &'a dyn TreeReader,
134 pub host: &'a dyn ApplyHost,
136 pub deploy: &'a dyn DeployApi,
138 pub gateway: &'a dyn GatewayApi,
140 pub backup: Option<&'a BackupNode<'a>>,
142 pub clock: &'a dyn Clock,
144 pub observer: &'a dyn ApplyObserver,
146}
147
148#[derive(Debug, Clone)]
150pub struct RollbackOptions {
151 pub run_id: RunId,
154 pub discard_live_changes: bool,
156 pub designers: DesignersPolicy,
158 pub hold: HoldModel,
160 pub poll: PollConfig,
162 pub backup: BackupPolicy,
165}
166
167#[derive(Debug, Clone, PartialEq, Eq)]
170pub struct RollbackChecked {
171 pub record: Option<RunRecord>,
175 pub live_changed: Vec<String>,
178 pub interrupted: Option<InterruptedRun>,
181}
182
183#[derive(Debug, Clone, PartialEq, Eq)]
185pub struct InterruptedRun {
186 pub journal: Journal,
188 pub live: BTreeMap<String, Option<String>>,
192 pub action: RecoveryAction,
194}
195
196#[derive(Debug, Clone, PartialEq, Serialize)]
198pub struct RollbackOutcome {
199 #[serde(rename = "outcome")]
201 pub kind: RollbackKind,
202 pub run_id: String,
204 pub target_run_id: String,
206 pub run_dir: String,
208 pub restored: Vec<String>,
210 pub deleted: Vec<String>,
212 pub discarded_live_changes: Vec<String>,
214 pub warnings: Vec<String>,
216 pub restart_pending: Vec<String>,
218 pub restart_pending_new: Vec<String>,
220 pub snapshot_sha256: BTreeMap<String, String>,
222 pub backup: Option<BackupReport>,
225}
226
227fn budget_files(record: &RunRecord) -> usize {
232 let (restored, deleted) = restore_set(record);
233 restored.len() + deleted.len()
234}
235
236pub fn restore_set(record: &RunRecord) -> (Vec<String>, Vec<String>) {
239 let mut restored = Vec::new();
240 let mut deleted = Vec::new();
241 for entry in &record.touched {
242 if entry.path == LEDGER_KEY {
243 continue;
244 }
245 if entry.pre_sha256.is_some() {
246 restored.push(entry.path.clone());
247 } else {
248 deleted.push(entry.path.clone());
249 }
250 }
251 (restored, deleted)
252}
253
254fn invalid(run_id: &str, problem: impl Into<String>) -> CoreError {
255 CoreError::DeployRunInvalid {
256 run_id: run_id.to_string(),
257 problem: problem.into(),
258 }
259}
260
261pub async fn check_rollback(
266 env: &str,
267 repo: &Path,
268 target: &Target,
269 reader: &dyn TreeReader,
270 host: &dyn ApplyHost,
271 run_id: &RunId,
272 discard_live_changes: bool,
273) -> Result<RollbackChecked, CoreError> {
274 let (gateway, mode) = (target.gateway.as_str(), target.mode.as_str());
275 let id = run_id.as_str();
276
277 let master = &target.nodes.master;
280 if let Some(journal) = journal_on_box(reader, master, gateway, mode).await? {
281 if journal.run_id != id {
282 return Err(journal_present(&journal, env, master, gateway, mode));
283 }
284 return check_interrupted(repo, target, host, run_id, journal).await;
285 }
286
287 let Some(record) = read_run_record(host, run_id).await? else {
289 return Err(invalid(
290 id,
291 format!("the host has no {RECORD_FILE} for this run (snapshots/{id}/{RECORD_FILE})"),
292 ));
293 };
294 validate_record(&record, target, run_id)?;
295 if let Some(refusal) = pair_tags_restore_refusal(&record, target) {
296 return Err(refusal);
297 }
298
299 cross_check_local(repo, &record)?;
301
302 let live_changed = live_changes(host, &record).await?;
304 if !live_changed.is_empty() && !discard_live_changes {
305 return Err(CoreError::DeployLiveChanged {
306 run_id: id.to_string(),
307 paths: live_changed
308 .iter()
309 .take(APPLY_LIST_SHOWN)
310 .cloned()
311 .collect(),
312 count: live_changed.len(),
313 });
314 }
315 Ok(RollbackChecked {
316 record: Some(record),
317 live_changed,
318 interrupted: None,
319 })
320}
321
322fn pair_tags_restore_refusal(record: &RunRecord, target: &Target) -> Option<CoreError> {
338 target.nodes.backup.as_ref()?;
340 let mut keys: Vec<String> = record
341 .touched
342 .iter()
343 .filter(|e| e.pre_sha256.is_some() && is_pair_unsynced_tag_path(&e.path))
344 .map(|e| {
345 e.path
346 .rsplit_once('/')
347 .map_or_else(|| e.path.clone(), |(dir, _)| dir.to_string())
348 })
349 .collect();
350 if keys.is_empty() {
351 return None;
352 }
353 keys.sort_by(|a, b| a.as_bytes().cmp(b.as_bytes()));
354 keys.dedup();
355 Some(CoreError::DeployPairTagsRefused {
356 mode: record.mode.clone(),
357 keys,
358 })
359}
360
361pub(crate) async fn live_changes(
364 host: &dyn ApplyHost,
365 record: &RunRecord,
366) -> Result<Vec<String>, CoreError> {
367 let touched: Vec<String> = record.touched.iter().map(|e| e.path.clone()).collect();
368 let live = host.fingerprint(&touched).await?;
369 Ok(record
370 .post_map()
371 .into_iter()
372 .filter(|(path, hash)| live.get(path) != Some(hash))
373 .map(|(path, _)| path)
374 .collect())
375}
376
377async fn read_run_record(
380 host: &dyn ApplyHost,
381 run_id: &RunId,
382) -> Result<Option<RunRecord>, CoreError> {
383 host.read_record(run_id).await.map_err(|e| match e {
384 CoreError::DeployRunInvalid { problem, .. } => invalid(run_id.as_str(), problem),
385 other => other,
386 })
387}
388
389async fn check_interrupted(
397 repo: &Path,
398 target: &Target,
399 host: &dyn ApplyHost,
400 run_id: &RunId,
401 journal: Journal,
402) -> Result<RollbackChecked, CoreError> {
403 let id = run_id.as_str();
404 let record = read_run_record(host, run_id).await?;
405 if let Some(record) = &record {
406 validate_identity(record, target, run_id)?;
407 validate_pre(record)?;
408 if journal.pre_fingerprint != record.pre_fingerprint
409 || journal.touched
410 != record
411 .touched
412 .iter()
413 .map(|e| e.path.clone())
414 .collect::<Vec<_>>()
415 {
416 return Err(invalid(
417 id,
418 "the journal and the record do not describe the same run",
419 ));
420 }
421 cross_check_local(repo, record)?;
422 }
423 let live = host.fingerprint(&journal.touched).await?;
424 let action = recovery_action(&journal, &live);
425 Ok(RollbackChecked {
426 record,
427 live_changed: Vec::new(),
428 interrupted: Some(InterruptedRun {
429 journal,
430 live,
431 action,
432 }),
433 })
434}
435
436fn validate_identity(record: &RunRecord, target: &Target, run_id: &RunId) -> Result<(), CoreError> {
438 let id = run_id.as_str();
439 if record.run_id != id {
440 return Err(invalid(id, "the record names a different run"));
441 }
442 if record.gateway != target.gateway {
443 return Err(invalid(
444 id,
445 "the record is for another gateway than the target's",
446 ));
447 }
448 if record.mode != target.mode {
449 return Err(invalid(
450 id,
451 "the record is for another mode than the target's",
452 ));
453 }
454 Ok(())
455}
456
457fn validate_pre(record: &RunRecord) -> Result<(), CoreError> {
459 if record.pre_fingerprint != fingerprint(&record.pre_map()) {
460 return Err(invalid(
461 &record.run_id,
462 "the record's pre_fingerprint is not the fingerprint of its touched set",
463 ));
464 }
465 Ok(())
466}
467
468fn validate_record(record: &RunRecord, target: &Target, run_id: &RunId) -> Result<(), CoreError> {
471 let id = run_id.as_str();
472 validate_identity(record, target, run_id)?;
473 match record.state {
474 RecordState::Applied => {}
475 RecordState::Applying => {
476 return Err(invalid(
477 id,
478 "the run did not finish and no journal is on the box, so there is nothing \
479 consistent to restore from; restore from the .gwbk backup or the \
480 run-directory snapshot",
481 ));
482 }
483 RecordState::RolledBack => {
484 let by = record.rolled_back_by.as_deref().unwrap_or("an unknown run");
485 return Err(invalid(id, format!("already rolled back by {by}")));
486 }
487 }
488 validate_pre(record)?;
489 if record.post_fingerprint.as_deref() != Some(fingerprint(&record.post_map()).as_str()) {
490 return Err(invalid(
491 id,
492 "the record's post_fingerprint is missing or is not the fingerprint of its touched set",
493 ));
494 }
495 Ok(())
496}
497
498fn read_local_json(
502 dir: &Path,
503 name: &str,
504 rel: &str,
505 id: &str,
506) -> Result<Option<Value>, CoreError> {
507 let path = dir.join(name);
508 let meta = match std::fs::symlink_metadata(&path) {
509 Ok(meta) => meta,
510 Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
511 Err(_) => return Err(invalid(id, format!("{rel}/{name} cannot be inspected"))),
512 };
513 if !meta.file_type().is_file() {
514 return Err(invalid(
515 id,
516 format!("{rel}/{name} is not a regular file (a link is not followed)"),
517 ));
518 }
519 let mut bytes = Vec::new();
520 std::fs::File::open(&path)
521 .and_then(|file| file.take(LOCAL_FILE_MAX_BYTES + 1).read_to_end(&mut bytes))
522 .map_err(|_| invalid(id, format!("{rel}/{name} cannot be read")))?;
523 if bytes.len() as u64 > LOCAL_FILE_MAX_BYTES {
524 return Err(invalid(id, format!("{rel}/{name} is over the size cap")));
525 }
526 serde_json::from_slice(&bytes)
527 .map(Some)
528 .map_err(|_| invalid(id, format!("{rel}/{name} is not valid JSON")))
529}
530
531fn cross_check_local(repo: &Path, record: &RunRecord) -> Result<(), CoreError> {
535 let id = record.run_id.as_str();
536 let run_id = RunId::parse(id)?;
537 if component_problem(&record.gateway).is_some() || record.gateway.contains(':') {
538 return Err(invalid(
539 id,
540 "the record's gateway is not a plain directory name",
541 ));
542 }
543 let rel = run_dir_rel(&record.gateway, &run_id);
544 let dir = rel
545 .split('/')
546 .fold(repo.to_path_buf(), |acc, c| acc.join(c));
547 match std::fs::symlink_metadata(&dir) {
548 Ok(meta) if meta.file_type().is_dir() => {}
549 Ok(_) => {
550 return Err(invalid(
551 id,
552 format!("{rel} is not a plain directory (a link is not followed)"),
553 ));
554 }
555 Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(()),
556 Err(_) => return Err(invalid(id, format!("{rel} cannot be inspected"))),
557 }
558
559 if let Some(run) = read_local_json(&dir, "run.json", &rel, id)?
564 && run.get("project").is_some_and(Value::is_string)
565 && run.get("mode").is_none()
566 {
567 let differs = |name: &str, field: &str| {
568 invalid(
569 id,
570 format!("{rel}/{name} field `{field}` differs from the on-box record"),
571 )
572 };
573 for (field, expected) in [("run_id", id), ("gateway", record.gateway.as_str())] {
574 if run.get(field).and_then(Value::as_str) != Some(expected) {
575 return Err(differs("run.json", field));
576 }
577 }
578 if let Some(plan) = read_local_json(&dir, "plan.json", &rel, id)? {
579 if plan.get("gateway").and_then(Value::as_str) != Some(record.gateway.as_str()) {
580 return Err(differs("plan.json", "gateway"));
581 }
582 if plan.pointer("/tags/plan_sha256").and_then(Value::as_str)
583 != Some(record.plan_sha256.as_str())
584 {
585 return Err(differs("plan.json", "tags.plan_sha256"));
586 }
587 }
588 return Ok(());
589 }
590
591 let on_box: [(&str, &str); 4] = [
592 ("gateway", &record.gateway),
593 ("mode", &record.mode),
594 ("plan_sha256", &record.plan_sha256),
595 ("render_sha256", &record.render_sha256),
596 ];
597 let compare = |name: &str, with_run_id: bool| -> Result<(), CoreError> {
598 let Some(doc) = read_local_json(&dir, name, &rel, id)? else {
599 return Ok(());
600 };
601 if name == "run.json" && doc.get("kind").and_then(Value::as_str) == Some("rollback") {
602 return Err(invalid(
603 id,
604 format!("{rel}/{name} is the run directory of a rollback, not of a deployment"),
605 ));
606 }
607 let mut fields: Vec<(&str, &str)> = on_box.to_vec();
608 if with_run_id {
609 fields.insert(0, ("run_id", id));
610 }
611 for (field, expected) in fields {
612 if doc.get(field).and_then(Value::as_str) != Some(expected) {
613 return Err(invalid(
614 id,
615 format!("{rel}/{name} field `{field}` differs from the on-box record"),
616 ));
617 }
618 }
619 Ok(())
620 };
621 compare("run.json", true)?;
622 compare("plan.json", false)
623}
624
625pub async fn run_rollback(
632 ctx: &RollbackContext<'_>,
633 opts: &RollbackOptions,
634) -> Result<RollbackOutcome, CoreError> {
635 let checked = check_rollback(
636 ctx.env,
637 ctx.repo,
638 ctx.target,
639 ctx.reader,
640 ctx.host,
641 &opts.run_id,
642 opts.discard_live_changes,
643 )
644 .await?;
645 let (run_id, run_dir) = RunDir::create_fresh(ctx.repo, &ctx.target.gateway)?;
646 let mut steps = Steps::new(None, ctx.observer);
647 let result = drive(ctx, opts, &checked, &run_id, &run_dir, &mut steps).await;
648
649 if let Err(e) = &result {
651 let _ = steps.emit(ApplyEvent::Failed {
652 code: e.code().to_string(),
653 });
654 }
655 let doc = match &result {
656 Ok(outcome) => {
657 let mut doc = serde_json::to_value(outcome).unwrap_or(Value::Null);
658 doc["ok"] = Value::Bool(true);
659 doc["kind"] = Value::String("rollback".to_string());
660 doc
661 }
662 Err(e) => json!({
663 "ok": false,
664 "kind": "rollback",
665 "run_id": run_id.as_str(),
666 "target_run_id": opts.run_id.as_str(),
667 "error": {"code": e.code(), "message": e.to_string()},
668 }),
669 };
670 let written = run_dir.write_json("result.json", &doc);
671 match result {
672 Ok(mut outcome) => {
673 if written.is_err() {
674 outcome.warnings.push("result_json_unwritable".to_string());
675 }
676 Ok(outcome)
677 }
678 Err(e) => Err(e),
679 }
680}
681
682async fn drive(
683 ctx: &RollbackContext<'_>,
684 opts: &RollbackOptions,
685 checked: &RollbackChecked,
686 run_id: &RunId,
687 run_dir: &RunDir,
688 steps: &mut Steps<'_>,
689) -> Result<RollbackOutcome, CoreError> {
690 let target = ctx.target;
691
692 let started_at = rfc3339_now_utc();
693 steps.log = Some(EventLog::open(run_dir)?);
694 let (writes, deletes) = checked.record.as_ref().map(restore_set).unwrap_or_default();
697 steps.emit(ApplyEvent::RunStarted {
698 run_id: run_id.as_str().to_string(),
699 writes: writes.len(),
700 deletes: deletes.len(),
701 })?;
702 let interrupted = checked.interrupted.as_ref();
703 if let Some(found) = interrupted {
704 steps.emit(ApplyEvent::InterruptedRun {
706 phase: found.journal.phase.as_str(),
707 action: found.action.as_str(),
708 hashes: found.live.clone(),
709 })?;
710 if found.action == RecoveryAction::ClearWithoutRestore {
711 return clear_run(ctx, opts, checked, run_id, run_dir, steps, &started_at).await;
712 }
713 }
714
715 let record = match (interrupted, &checked.record) {
721 (Some(found), record) => {
722 snapshot_record(ctx, opts, run_dir, found, record.as_ref()).await?
723 }
724 (None, Some(record)) => record.clone(),
725 (None, None) => {
726 return Err(CoreError::Internal(
727 "a rollback of a finished run has no record".to_string(),
728 ));
729 }
730 };
731 let record = &record;
732 let (restored, deleted) = restore_set(record);
733
734 let touched_paths: Vec<String> = record.touched.iter().map(|e| e.path.clone()).collect();
735 let preflights = run_preflights(
736 &PreflightInput {
737 target,
738 declared: None,
739 master: ctx.deploy,
740 backup: ctx.backup,
741 gateway: ctx.gateway,
742 reader: ctx.reader,
743 host: ctx.host,
744 touched_files: budget_files(record),
745 touched_paths: &touched_paths,
746 deleted_paths: &[],
749 hold: &opts.hold,
750 designers: &opts.designers,
751 clock: ctx.clock,
752 },
753 steps,
754 )
755 .await;
756 let seen = preflights.as_ref().ok();
757 run_dir.write_json(
758 "run.json",
759 &json!({
760 "schema_version": 1,
761 "kind": "rollback",
762 "run_id": run_id.as_str(),
763 "target_run_id": opts.run_id.as_str(),
764 "env": ctx.env,
765 "target_file": ctx.target_file,
766 "gateway": target.gateway,
767 "mode": target.mode,
768 "node": "master",
769 "started_at": started_at,
770 "run_dir_git": run_dir.git_status().as_record(),
771 "render_sha256": record.render_sha256,
772 "plan_sha256": record.plan_sha256,
773 "hold_s": opts.hold.hold_s,
774 "discard_live_changes": opts.discard_live_changes,
775 "discarded_live_changes": checked.live_changed,
776 "interrupted_phase": interrupted.map(|i| i.journal.phase.as_str()),
777 "recovery": interrupted.map(|i| i.action.as_str()),
778 "designers": opts.designers.mode.as_str(),
779 "designers_timeout_s": opts.designers.timeout.as_secs(),
780 "designers_seen": seen.map(|r| r.designers_seen
781 .iter()
782 .map(|(user, project, host)| json!({"user": user, "project": project, "host": host}))
783 .collect::<Vec<_>>()),
784 "preflights": seen.map(|r| r.findings
785 .iter()
786 .map(|f| json!({"check": f.check.as_str(), "status": f.status.as_str()}))
787 .collect::<Vec<_>>()),
788 }),
789 )?;
790 run_dir.write_json(
791 "plan.json",
792 &json!({
793 "schema_version": 1,
794 "kind": "rollback",
795 "target_run_id": opts.run_id.as_str(),
796 "gateway": target.gateway,
797 "mode": target.mode,
798 "plan_sha256": record.plan_sha256,
799 "render_sha256": record.render_sha256,
800 "restore": restored,
801 "delete": deleted,
802 "ledger": record
803 .touched
804 .iter()
805 .find(|e| e.path == LEDGER_KEY)
806 .map(|e| if e.pre_sha256.is_some() { "restore" } else { "delete" }),
807 }),
808 )?;
809 let preflights = preflights?;
810 if !preflights.passed() {
811 steps.emit(ApplyEvent::PreflightChecks {
812 checks: preflights
813 .findings
814 .iter()
815 .map(|f| (f.check.as_str().to_string(), f.status.as_str().to_string()))
816 .collect(),
817 })?;
818 return Err(CoreError::DeployPreflightFailed {
819 failures: preflights.failures().into(),
820 });
821 }
822 if !checked.live_changed.is_empty() {
823 steps.emit(ApplyEvent::LiveChangesDiscarded {
824 count: checked.live_changed.len(),
825 paths: checked
826 .live_changed
827 .iter()
828 .take(APPLY_LIST_SHOWN)
829 .cloned()
830 .collect(),
831 })?;
832 }
833
834 let restart_before = ctx.deploy.restart_pending().await.ok();
837
838 let lock = HeldLock::acquire(ctx.deploy, ScanKind::Config, opts.hold.hold_s, ctx.clock).await?;
842 steps.emit(ApplyEvent::LockAcquired {
843 hold_s: opts.hold.hold_s,
844 })?;
845
846 let mut live_changed = checked.live_changed.clone();
853 if interrupted.is_none() {
854 let now = live_changes(ctx.host, record).await?;
855 if now != live_changed {
856 if !opts.discard_live_changes {
857 let _ = ctx.deploy.scan_request(ScanKind::Config).await;
860 return Err(CoreError::DeployLiveChanged {
861 run_id: opts.run_id.as_str().to_string(),
862 paths: now.iter().take(APPLY_LIST_SHOWN).cloned().collect(),
863 count: now.len(),
864 });
865 }
866 steps.emit(ApplyEvent::LiveChangesDiscarded {
867 count: now.len(),
868 paths: now.iter().take(APPLY_LIST_SHOWN).cloned().collect(),
869 })?;
870 live_changed = now;
871 }
872 }
873 let backup_baseline = match ctx.backup {
875 Some(backup) => Some(restore_baseline(backup).await),
876 None => None,
877 };
878 let report = restore_snapshot(
879 &RestoreContext {
880 env: ctx.env,
881 master: &target.nodes.master,
882 host: ctx.host,
883 deploy: ctx.deploy,
884 clock: ctx.clock,
885 hold: &opts.hold,
886 poll: &opts.poll,
887 },
888 &RestoreInput {
889 run_id,
890 run_dir,
891 record,
892 held: Some(&lock),
893 cause: None,
894 },
895 steps,
896 )
897 .await?;
898
899 let mut warnings = report.warnings;
900 let (restart_pending, restart_pending_new) =
901 restart_report(ctx.deploy, restart_before.as_deref(), &mut warnings).await;
902 let backup = match (ctx.backup, backup_baseline) {
904 (Some(backup), Some(baseline)) => Some(
905 restore_backup_phase(
906 RestoreSurfaces {
907 deploy: ctx.deploy,
908 reader: ctx.reader,
909 clock: ctx.clock,
910 backup,
911 },
912 record,
913 &target.mode,
914 opts.backup,
915 baseline,
916 steps,
917 &mut warnings,
918 )
919 .await,
920 ),
921 _ => None,
922 };
923 Ok(RollbackOutcome {
924 kind: RollbackKind::Restored,
925 run_id: run_id.as_str().to_string(),
926 target_run_id: opts.run_id.as_str().to_string(),
927 run_dir: run_dir.path().to_string_lossy().into_owned(),
928 restored,
929 deleted,
930 discarded_live_changes: live_changed,
931 warnings,
932 restart_pending,
933 restart_pending_new,
934 snapshot_sha256: report.snapshot_sha256,
935 backup,
936 })
937}
938
939const UNKNOWN_SHA256: &str = "0000000000000000000000000000000000000000000000000000000000000000";
942
943fn snapshot_key(path: &str) -> String {
947 if path == LEDGER_KEY {
948 LEDGER_KEY.to_string()
949 } else {
950 super::snapshot_slot(path)
951 }
952}
953
954fn named(paths: &[String]) -> String {
956 let shown = paths.len().min(APPLY_LIST_SHOWN);
957 let mut out = paths[..shown].join(", ");
958 if paths.len() > shown {
959 out.push_str(&format!(" and {} more", paths.len() - shown));
960 }
961 out
962}
963
964async fn snapshot_record(
975 ctx: &RollbackContext<'_>,
976 opts: &RollbackOptions,
977 run_dir: &RunDir,
978 found: &InterruptedRun,
979 record: Option<&RunRecord>,
980) -> Result<RunRecord, CoreError> {
981 let target = ctx.target;
982 let (gateway, mode) = (target.gateway.as_str(), target.mode.as_str());
983 let fail = |cause: String| {
984 CoreError::RollbackFailed(Box::new(RollbackFailure {
985 run_id: opts.run_id.as_str().to_string(),
986 env: ctx.env.to_string(),
987 run_dir: Some(run_dir.path().to_string_lossy().into_owned()),
988 journal: journal_location(LedgerRoot::for_node(&target.nodes.master), gateway, mode),
989 cause,
990 scope: RollbackScope::Files,
991 }))
992 };
993
994 let scratch = run_dir.path().join("inspect");
995 std::fs::create_dir(&scratch)
996 .map_err(|e| fail(format!("cannot read the snapshot ({:?})", e.kind())))?;
997 let fetched = ctx.host.fetch_snapshot(&opts.run_id, &scratch).await;
998 let _ = std::fs::remove_dir_all(&scratch);
1001 let hashes = fetched.map_err(|e| fail(format!("the snapshot cannot be read: {e}")))?;
1002
1003 if let Some(record) = record {
1004 let bad: Vec<String> = record
1005 .touched
1006 .iter()
1007 .filter_map(|entry| {
1008 let pre = entry.pre_sha256.as_deref()?;
1009 (hashes.get(&snapshot_key(&entry.path)).map(String::as_str) != Some(pre))
1010 .then(|| entry.path.clone())
1011 })
1012 .collect();
1013 if !bad.is_empty() {
1014 return Err(fail(format!(
1015 "the snapshot is missing or damaged for {} path(s): {}",
1016 bad.len(),
1017 named(&bad)
1018 )));
1019 }
1020 return Ok(record.clone());
1021 }
1022
1023 let pre: BTreeMap<String, Option<String>> = found
1024 .journal
1025 .touched
1026 .iter()
1027 .map(|path| (path.clone(), hashes.get(&snapshot_key(path)).cloned()))
1028 .collect();
1029 if fingerprint(&pre) != found.journal.pre_fingerprint {
1030 return Err(fail(
1031 "the run's record is missing and the snapshot is incomplete: its files do not \
1032 match the journal's pre-run fingerprint"
1033 .to_string(),
1034 ));
1035 }
1036 Ok(RunRecord {
1037 schema_version: 1,
1038 run_id: found.journal.run_id.clone(),
1039 gateway: found.journal.gateway.clone(),
1040 mode: found.journal.mode.clone(),
1041 render_sha256: UNKNOWN_SHA256.to_string(),
1042 plan_sha256: UNKNOWN_SHA256.to_string(),
1043 state: RecordState::Applying,
1044 rolled_back_by: None,
1045 touched: found
1046 .journal
1047 .touched
1048 .iter()
1049 .map(|path| TouchedEntry {
1050 path: path.clone(),
1051 pre_sha256: pre.get(path).cloned().flatten(),
1052 post_sha256: None,
1053 })
1054 .collect(),
1055 pre_fingerprint: found.journal.pre_fingerprint.clone(),
1056 post_fingerprint: None,
1057 })
1058}
1059
1060async fn clear_run(
1064 ctx: &RollbackContext<'_>,
1065 opts: &RollbackOptions,
1066 checked: &RollbackChecked,
1067 run_id: &RunId,
1068 run_dir: &RunDir,
1069 steps: &mut Steps<'_>,
1070 started_at: &str,
1071) -> Result<RollbackOutcome, CoreError> {
1072 let target = ctx.target;
1073 let phase = checked
1074 .interrupted
1075 .as_ref()
1076 .map(|i| i.journal.phase.as_str());
1077 run_dir.write_json(
1078 "run.json",
1079 &json!({
1080 "schema_version": 1,
1081 "kind": "rollback",
1082 "run_id": run_id.as_str(),
1083 "target_run_id": opts.run_id.as_str(),
1084 "env": ctx.env,
1085 "target_file": ctx.target_file,
1086 "gateway": target.gateway,
1087 "mode": target.mode,
1088 "node": "master",
1089 "started_at": started_at,
1090 "run_dir_git": run_dir.git_status().as_record(),
1091 "discard_live_changes": opts.discard_live_changes,
1092 "discarded_live_changes": checked.live_changed,
1093 "interrupted_phase": phase,
1094 "recovery": RecoveryAction::ClearWithoutRestore.as_str(),
1095 }),
1096 )?;
1097 run_dir.write_json(
1098 "plan.json",
1099 &json!({
1100 "schema_version": 1,
1101 "kind": "rollback",
1102 "target_run_id": opts.run_id.as_str(),
1103 "gateway": target.gateway,
1104 "mode": target.mode,
1105 "restore": Vec::<String>::new(),
1106 "delete": Vec::<String>::new(),
1107 "ledger": Value::Null,
1108 "recovery": RecoveryAction::ClearWithoutRestore.as_str(),
1109 }),
1110 )?;
1111 ctx.host.discard_interrupted(&opts.run_id).await?;
1112 steps.emit(ApplyEvent::Cleared)?;
1113 Ok(RollbackOutcome {
1114 kind: RollbackKind::Cleared,
1115 run_id: run_id.as_str().to_string(),
1116 target_run_id: opts.run_id.as_str().to_string(),
1117 run_dir: run_dir.path().to_string_lossy().into_owned(),
1118 restored: Vec::new(),
1119 deleted: Vec::new(),
1120 discarded_live_changes: Vec::new(),
1121 warnings: Vec::new(),
1122 restart_pending: Vec::new(),
1123 restart_pending_new: Vec::new(),
1124 snapshot_sha256: BTreeMap::new(),
1125 backup: None,
1126 })
1127}
1128
1129#[cfg(test)]
1130mod tests {
1131 use super::*;
1132 use crate::deploy::apply::preflight::{FindingStatus, hold_budget_finding};
1133
1134 fn record_of(files: usize) -> RunRecord {
1136 let mut touched: Vec<TouchedEntry> = (0..files)
1137 .map(|i| TouchedEntry {
1138 path: format!("u{i}/config.json"),
1139 pre_sha256: Some("a".repeat(64)),
1140 post_sha256: Some("b".repeat(64)),
1141 })
1142 .collect();
1143 touched.push(TouchedEntry {
1144 path: LEDGER_KEY.to_string(),
1145 pre_sha256: Some("c".repeat(64)),
1146 post_sha256: Some("d".repeat(64)),
1147 });
1148 RunRecord {
1149 schema_version: 1,
1150 run_id: "20261016T000000Z-aaaaaa".to_string(),
1151 gateway: "gw".to_string(),
1152 mode: "dev".to_string(),
1153 render_sha256: "e".repeat(64),
1154 plan_sha256: "f".repeat(64),
1155 state: RecordState::Applied,
1156 rolled_back_by: None,
1157 touched,
1158 pre_fingerprint: "0".repeat(64),
1159 post_fingerprint: None,
1160 }
1161 }
1162
1163 fn rollback_budget(files: usize) -> FindingStatus {
1165 let record = record_of(files);
1166 hold_budget_finding(&HoldModel::MEASURED, budget_files(&record)).status
1167 }
1168
1169 #[test]
1170 fn snapshot_key_core_key_reads_the_core_slot() {
1171 assert_eq!(
1172 snapshot_key(".core/ignition/tag-definition/default/Deploy/tags.json"),
1173 "core/ignition/tag-definition/default/Deploy/tags.json"
1174 );
1175 assert_eq!(
1176 snapshot_key("ignition/tag-definition/default/Deploy/tags.json"),
1177 "files/ignition/tag-definition/default/Deploy/tags.json"
1178 );
1179 assert_eq!(snapshot_key(LEDGER_KEY), LEDGER_KEY);
1180 }
1181
1182 #[test]
1183 fn rollback_budget_counts_what_the_apply_counted() {
1184 assert_eq!(record_of(1527).touched.len(), 1528);
1188 assert_eq!(budget_files(&record_of(1527)), 1527);
1189 assert_eq!(rollback_budget(1527), FindingStatus::Pass);
1190 assert_eq!(budget_files(&record_of(1528)), 1528);
1192 assert_eq!(rollback_budget(1528), FindingStatus::Fail);
1193 }
1194
1195 fn target(pair: bool) -> Target {
1196 let mut yaml = "schema_version: 1\ngateway: gw\nmode: dev\nnodes:\n master:\n profile: a\n local:\n data_dir: /data/a\n".to_string();
1197 if pair {
1198 yaml.push_str(" backup:\n profile: b\n local:\n data_dir: /data/b\n");
1199 }
1200 Target::parse("targets/dev.yaml", yaml.as_bytes()).expect("target parses")
1201 }
1202
1203 fn touched(path: &str, pre: bool, post: bool) -> TouchedEntry {
1204 TouchedEntry {
1205 path: path.to_string(),
1206 pre_sha256: pre.then(|| "a".repeat(64)),
1207 post_sha256: post.then(|| "b".repeat(64)),
1208 }
1209 }
1210
1211 #[test]
1217 fn pair_rollback_refuses_restoring_tag_files_into_the_mode_collection() {
1218 let mut record = record_of(1);
1219 record.touched = vec![
1220 touched(
1221 "ignition/tag-definition/default/Deploy/tags.json",
1222 true,
1223 false,
1224 ),
1225 touched(
1226 "ignition/tag-definition/default/Deploy/unary-resource.json",
1227 true,
1228 false,
1229 ),
1230 touched(
1231 "ignition/tag-type-definition/default/T/udts.json",
1232 true,
1233 true,
1234 ),
1235 touched("ignition/tag-group/default/Fast/config.json", true, true),
1236 touched("ignition/tag-definition/default/New/tags.json", false, true),
1237 touched(LEDGER_KEY, true, true),
1238 ];
1239 let err = pair_tags_restore_refusal(&record, &target(true)).expect("refused");
1240 assert_eq!(err.code(), "deploy_pair_tags_refused");
1241 match err {
1242 CoreError::DeployPairTagsRefused { mode, keys } => {
1243 assert_eq!(mode, "dev");
1244 assert_eq!(
1245 keys,
1246 vec![
1247 "ignition/tag-definition/default/Deploy".to_string(),
1248 "ignition/tag-type-definition/default/T".to_string(),
1249 ]
1250 );
1251 }
1252 other => panic!("expected DeployPairTagsRefused, got {other:?}"),
1253 }
1254 assert!(pair_tags_restore_refusal(&record, &target(false)).is_none());
1255
1256 record.touched = vec![
1258 touched("ignition/tag-definition/default/New/tags.json", false, true),
1259 touched(
1260 "ignition/tag-definition/default/New/unary-resource.json",
1261 false,
1262 true,
1263 ),
1264 touched(
1265 "ignition/database-translator/POSTGRES/config.json",
1266 true,
1267 true,
1268 ),
1269 touched(LEDGER_KEY, false, true),
1270 ];
1271 assert!(pair_tags_restore_refusal(&record, &target(true)).is_none());
1272 }
1273
1274 #[test]
1278 fn rollback_core_only_run_allowed_on_pair() {
1279 let mut record = record_of(1);
1280 record.touched = vec![
1281 touched(
1282 ".core/ignition/tag-definition/default/Deploy/tags.json",
1283 true,
1284 true,
1285 ),
1286 touched(
1287 ".core/ignition/tag-definition/default/Deploy/unary-resource.json",
1288 false,
1289 true,
1290 ),
1291 touched(
1292 ".core/ignition/tag-type-definition/default/T/udts.json",
1293 true,
1294 false,
1295 ),
1296 touched(LEDGER_KEY, true, true),
1297 ];
1298 assert!(pair_tags_restore_refusal(&record, &target(true)).is_none());
1299 }
1300
1301 #[test]
1305 fn rollback_migration_run_refused_on_pair() {
1306 let mut record = record_of(1);
1307 record.touched = vec![
1308 touched(
1309 ".core/ignition/tag-definition/default/Deploy/tags.json",
1310 false,
1311 true,
1312 ),
1313 touched(
1314 ".core/ignition/tag-definition/default/Deploy/unary-resource.json",
1315 false,
1316 true,
1317 ),
1318 touched(
1319 "ignition/tag-definition/default/Deploy/tags.json",
1320 true,
1321 false,
1322 ),
1323 touched(
1324 "ignition/tag-definition/default/Deploy/unary-resource.json",
1325 true,
1326 false,
1327 ),
1328 touched(LEDGER_KEY, true, true),
1329 ];
1330 let err = pair_tags_restore_refusal(&record, &target(true)).expect("refused");
1331 assert_eq!(err.code(), "deploy_pair_tags_refused");
1332 match &err {
1333 CoreError::DeployPairTagsRefused { keys, .. } => {
1334 assert_eq!(
1335 keys,
1336 &vec!["ignition/tag-definition/default/Deploy".to_string()]
1337 );
1338 }
1339 other => panic!("expected DeployPairTagsRefused, got {other:?}"),
1340 }
1341 let hint = err.hint().expect("hint");
1342 assert!(hint.contains("apply again"), "{hint}");
1343 assert!(pair_tags_restore_refusal(&record, &target(false)).is_none());
1344 }
1345
1346 fn project_run_dir(record: &RunRecord, plan_sha256: &str) -> tempfile::TempDir {
1351 let repo = tempfile::tempdir().expect("tempdir");
1352 let run_id = RunId::parse(&record.run_id).expect("run id");
1353 let dir = run_dir_rel(&record.gateway, &run_id)
1354 .split('/')
1355 .fold(repo.path().to_path_buf(), |acc, c| acc.join(c));
1356 std::fs::create_dir_all(&dir).expect("run dir");
1357 let run = serde_json::json!({
1358 "run_id": record.run_id, "env": "dev", "gateway": record.gateway,
1359 "project": "CapApp", "release": "v1.0.0", "action": "import",
1360 "tag_providers": ["PlantA"]
1361 });
1362 let plan = serde_json::json!({
1363 "env": "dev", "gateway": record.gateway, "project": "CapApp",
1364 "tags": {"plan_sha256": plan_sha256}
1365 });
1366 std::fs::write(dir.join("run.json"), run.to_string()).expect("run.json");
1367 std::fs::write(dir.join("plan.json"), plan.to_string()).expect("plan.json");
1368 repo
1369 }
1370
1371 #[test]
1372 fn a_project_run_directory_cross_checks_its_group_record() {
1373 let record = record_of(1);
1374 let repo = project_run_dir(&record, &record.plan_sha256);
1375 cross_check_local(repo.path(), &record).expect("the project run's own record");
1376 let other = project_run_dir(&record, &"9".repeat(64));
1377 let err = cross_check_local(other.path(), &record).expect_err("another plan");
1378 assert_eq!(err.code(), "deploy_run_invalid", "{err}");
1379 }
1380}