1use std::path::{Path, PathBuf};
9
10use serde::Serialize;
11
12use std::time::Duration;
13
14use crate::actions::deploy::{
15 Declared, DeployPlanResult, PlanOptions, VerifyOptions, deploy_plan, load_target,
16};
17use crate::client::{DeployApi, GatewayApi};
18use crate::deploy::apply::gates::{
19 GateFlags, GateOutcome, evaluate_gates, validate_drift_reason, validate_plan_sha256,
20};
21use crate::deploy::apply::hold::{HoldModel, SystemClock};
22use crate::deploy::apply::host::ApplyHost;
23use crate::deploy::apply::journal::RecoveryAction;
24use crate::deploy::apply::preflight::DesignersPolicy;
25use crate::deploy::apply::record::RunRecord;
26use crate::deploy::apply::run_store::RunId;
27use crate::deploy::apply::{
28 ApplyContext, ApplyOptions, BACKUP_UNSYNCED_SLUG, InterruptedRun, OutcomeKind,
29 ROLLED_BACK_SLUG, RollbackCause, RollbackChecked, RollbackContext, RollbackKind,
30 RollbackOptions, check_rollback, restore_set, run_apply, run_rollback,
31};
32use crate::deploy::apply::{ApplyObserver, ApplyOptions as EngineOptions};
33use crate::deploy::pair::{BACKUP_POLL_S, BackupNode, BackupPolicy, BackupReport};
34use crate::deploy::plan::PlanSummary;
35use crate::deploy::target::Target;
36use crate::deploy::transport::TreeReader;
37use crate::deploy::transport::apply_host_for_node;
38use crate::error::CoreError;
39use crate::poll::PollConfig;
40
41#[derive(Debug, Clone, Default)]
43pub struct PrepareOptions {
44 pub base: Option<PathBuf>,
46 pub gates: GateFlags,
48}
49
50#[derive(Debug, Clone, PartialEq)]
52pub enum ApplyPrepared {
53 Noop(DeployApplyResult),
55 Ready {
57 plan: DeployPlanResult,
59 prose: String,
61 },
62}
63
64#[derive(Debug, Clone, PartialEq, Serialize)]
66pub struct DeployApplyResult {
67 pub env: String,
69 pub gateway: String,
71 pub mode: String,
73 pub node: String,
75 pub transport: String,
77 pub run_id: Option<String>,
79 pub run_dir: Option<String>,
81 pub outcome: OutcomeKind,
83 pub ok: bool,
85 pub slug: Option<&'static str>,
89 pub cause: Option<RollbackCause>,
91 pub plan_sha256: String,
93 pub render_sha256: String,
95 pub summary: PlanSummary,
97 pub written: Vec<String>,
99 pub deleted: Vec<String>,
101 pub warnings: Vec<String>,
103 pub gwbk: Option<String>,
106 pub skip_gwbk: bool,
109 pub restart_pending: Vec<String>,
111 pub restart_pending_new: Vec<String>,
113 pub backup: Option<Box<BackupReport>>,
117}
118
119impl DeployApplyResult {
120 pub fn failure_exit_code(&self) -> Option<u8> {
123 matches!(
124 self.outcome,
125 OutcomeKind::RolledBack | OutcomeKind::AppliedBackupUnsynced
126 )
127 .then_some(6)
128 }
129}
130
131fn base_result(plan: &DeployPlanResult, outcome: OutcomeKind) -> DeployApplyResult {
133 DeployApplyResult {
134 env: plan.env.clone(),
135 gateway: plan.gateway.clone(),
136 mode: plan.mode.clone(),
137 node: plan.node.clone(),
138 transport: plan.transport.clone(),
139 run_id: None,
140 run_dir: None,
141 outcome,
142 ok: !matches!(
143 outcome,
144 OutcomeKind::RolledBack | OutcomeKind::AppliedBackupUnsynced
145 ),
146 slug: match outcome {
147 OutcomeKind::RolledBack => Some(ROLLED_BACK_SLUG),
148 OutcomeKind::AppliedBackupUnsynced => Some(BACKUP_UNSYNCED_SLUG),
149 OutcomeKind::Applied | OutcomeKind::Noop => None,
150 },
151 cause: None,
152 plan_sha256: plan.plan_sha256.clone(),
153 render_sha256: plan.render_sha256.clone(),
154 summary: plan.summary.clone(),
155 written: Vec::new(),
156 deleted: Vec::new(),
157 warnings: plan.warnings.clone(),
158 gwbk: None,
159 skip_gwbk: false,
160 restart_pending: Vec::new(),
161 restart_pending_new: Vec::new(),
162 backup: None,
163 }
164}
165
166pub async fn deploy_apply_prepare(
173 declared: &Declared,
174 reader: &dyn TreeReader,
175 opts: &PrepareOptions,
176) -> Result<ApplyPrepared, CoreError> {
177 if let Some(hash) = &opts.gates.plan_sha256 {
179 validate_plan_sha256(hash)?;
180 }
181 if let Some(reason) = &opts.gates.allow_drift {
182 validate_drift_reason(reason)?;
183 }
184 let plan = deploy_plan(
185 declared,
186 reader,
187 &PlanOptions {
188 base: opts.base.clone(),
189 diff: false,
190 },
191 )
192 .await?;
193 match evaluate_gates(&plan, &opts.gates)? {
194 GateOutcome::Noop => Ok(ApplyPrepared::Noop(base_result(&plan, OutcomeKind::Noop))),
195 GateOutcome::Ready { prose } => Ok(ApplyPrepared::Ready { plan, prose }),
196 }
197}
198
199pub async fn deploy_apply_execute(
202 prepared: ApplyPrepared,
203 ctx: &ApplyContext<'_>,
204 opts: &ApplyOptions,
205) -> Result<DeployApplyResult, CoreError> {
206 let plan = match prepared {
207 ApplyPrepared::Noop(result) => return Ok(result),
208 ApplyPrepared::Ready { plan, .. } => plan,
209 };
210 let outcome = run_apply(ctx, &plan, opts).await?;
211 let mut result = base_result(&plan, outcome.kind);
212 result.run_id = outcome.run_id;
213 result.run_dir = outcome.run_dir;
214 result.cause = outcome.cause;
215 result.written = outcome.written;
216 result.deleted = outcome.deleted;
217 result.warnings.extend(outcome.warnings);
218 result.gwbk = outcome.gwbk;
219 result.skip_gwbk = outcome.skip_gwbk;
220 result.restart_pending = outcome.restart_pending;
221 result.restart_pending_new = outcome.restart_pending_new;
222 result.backup = outcome.backup.map(Box::new);
223 Ok(result)
224}
225
226pub struct ApplySurfaces<'a> {
228 pub deploy: &'a dyn DeployApi,
230 pub backup: Option<&'a BackupNode<'a>>,
233 pub gateway: &'a dyn GatewayApi,
235}
236
237#[derive(Debug, Clone)]
239pub struct ApplyRunOptions {
240 pub base: Option<PathBuf>,
242 pub gates: GateFlags,
244 pub project: String,
246 pub designers: DesignersPolicy,
248 pub skip_gwbk: bool,
250 pub backup_timeout: Duration,
253}
254
255const SCAN_POLL_INTERVAL: Duration = Duration::from_millis(500);
257
258const SCAN_WAIT: Duration = Duration::from_secs(120);
261
262fn scan_poll(subject: &str) -> PollConfig {
264 PollConfig {
265 subject: subject.to_string(),
266 interval: SCAN_POLL_INTERVAL,
267 max: Duration::from_secs(2),
268 deadline: SCAN_WAIT,
269 }
270}
271
272pub async fn deploy_apply_run(
281 prepared: ApplyPrepared,
282 declared: &Declared,
283 repo: &Path,
284 reader: &dyn TreeReader,
285 surfaces: &ApplySurfaces<'_>,
286 opts: &ApplyRunOptions,
287 observer: &dyn ApplyObserver,
288) -> Result<DeployApplyResult, CoreError> {
289 deploy_apply_run_in(
290 prepared, declared, repo, repo, reader, surfaces, opts, observer,
291 )
292 .await
293}
294
295#[allow(clippy::too_many_arguments)]
299pub async fn deploy_apply_run_in(
300 prepared: ApplyPrepared,
301 declared: &Declared,
302 repo: &Path,
303 targets: &Path,
304 reader: &dyn TreeReader,
305 surfaces: &ApplySurfaces<'_>,
306 opts: &ApplyRunOptions,
307 observer: &dyn ApplyObserver,
308) -> Result<DeployApplyResult, CoreError> {
309 if let ApplyPrepared::Noop(result) = prepared {
310 return Ok(result);
311 }
312 let host = apply_host_for_node(
313 targets,
314 &declared.target_file,
315 "nodes.master",
316 &declared.target.nodes.master,
317 &declared.target.gateway,
318 &declared.target.mode,
319 )?;
320 let ctx = ApplyContext {
321 declared,
322 repo,
323 reader,
324 host: host.as_ref(),
325 deploy: surfaces.deploy,
326 gateway: surfaces.gateway,
327 backup: surfaces.backup,
328 clock: &SystemClock,
329 observer,
330 };
331 let engine = EngineOptions {
332 base: opts.base.clone(),
333 gates: opts.gates.clone(),
334 project: opts.project.clone(),
335 hold: HoldModel::MEASURED,
336 poll: scan_poll("config scan after apply"),
337 verify: VerifyOptions::default(),
338 designers: opts.designers,
339 skip_gwbk: opts.skip_gwbk,
340 backup: BackupPolicy {
341 budget: opts.backup_timeout,
342 interval: Duration::from_secs(BACKUP_POLL_S),
343 },
344 };
345 deploy_apply_execute(prepared, &ctx, &engine).await
346}
347
348#[derive(Debug, Clone, Copy, Default)]
354pub struct RollbackPrepareOptions {
355 pub discard_live_changes: bool,
357}
358
359#[derive(Debug, Clone)]
361pub struct RollbackPrepared {
362 pub run_id: String,
364 pub record: Option<RunRecord>,
368 pub interrupted: Option<InterruptedRun>,
371 pub prose: String,
373 pub live_changed: Vec<String>,
376 pub target: Target,
379 pub target_file: String,
381}
382
383#[derive(Debug, Clone, PartialEq, Serialize)]
385pub struct DeployRollbackResult {
386 pub env: String,
388 pub gateway: String,
390 pub mode: String,
392 pub run_id: String,
394 pub target_run_id: String,
396 pub run_dir: String,
398 pub outcome: RollbackKind,
400 pub ok: bool,
402 pub restored: Vec<String>,
404 pub deleted: Vec<String>,
406 pub discarded_live_changes: Vec<String>,
409 pub warnings: Vec<String>,
411 pub restart_pending: Vec<String>,
413 pub restart_pending_new: Vec<String>,
415}
416
417fn rollback_prose(run_id: &str, target: &Target, checked: &RollbackChecked) -> String {
419 let record = checked.record.as_ref();
420 let (restored, deleted) = record.map(restore_set).unwrap_or_default();
421 let ledger = record
422 .and_then(|r| {
423 r.touched
424 .iter()
425 .find(|e| e.path == crate::deploy::apply::fingerprint::LEDGER_KEY)
426 })
427 .map(|e| {
428 if e.pre_sha256.is_some() {
429 ", and restore the ledger"
430 } else {
431 ", and remove the ledger"
432 }
433 })
434 .unwrap_or("");
435 let mut prose = match (&checked.interrupted, record) {
436 (Some(found), _) if found.action == RecoveryAction::ClearWithoutRestore => format!(
437 "Clear run {run_id} on gateway {} (mode {}), which was interrupted at phase {} \
438 before any file moved and whose touched files are as they were before the run: \
439 remove its journal, partial snapshot and staging. Nothing is restored.",
440 target.gateway,
441 target.mode,
442 found.journal.phase.as_str(),
443 ),
444 (Some(found), Some(record)) => format!(
445 "Recover run {run_id} on gateway {} (mode {}), which was interrupted at phase {}: \
446 restore {} file(s) from the on-box snapshot and delete {} file(s) the run \
447 created{ledger}, then remove the journal. The run was applying plan_sha256 {}.",
448 target.gateway,
449 target.mode,
450 found.journal.phase.as_str(),
451 restored.len(),
452 deleted.len(),
453 record.plan_sha256,
454 ),
455 (Some(found), None) => format!(
456 "Recover run {run_id} on gateway {} (mode {}), which was interrupted at phase {} \
457 and whose record is missing: restore the touched files from the on-box snapshot, \
458 then remove the journal.",
459 target.gateway,
460 target.mode,
461 found.journal.phase.as_str(),
462 ),
463 (None, Some(record)) => format!(
464 "Roll back run {run_id} on gateway {} (mode {}): restore {} file(s) from the on-box \
465 snapshot and delete {} file(s) the run created{ledger}. The run applied \
466 plan_sha256 {}.",
467 target.gateway,
468 target.mode,
469 restored.len(),
470 deleted.len(),
471 record.plan_sha256,
472 ),
473 (None, None) => format!("Roll back run {run_id} on gateway {}.", target.gateway),
474 };
475 if !checked.live_changed.is_empty() {
476 prose.push_str(&format!(
477 " {} file(s) changed after the run and will be overwritten: {}.",
478 checked.live_changed.len(),
479 checked
480 .live_changed
481 .iter()
482 .take(crate::error::APPLY_LIST_SHOWN)
483 .cloned()
484 .collect::<Vec<_>>()
485 .join(", ")
486 ));
487 }
488 prose
489}
490
491pub async fn deploy_rollback_prepare(
500 repo: &Path,
501 env: &str,
502 run_id_text: &str,
503 reader: &dyn TreeReader,
504 host: &dyn ApplyHost,
505 opts: &RollbackPrepareOptions,
506) -> Result<RollbackPrepared, CoreError> {
507 deploy_rollback_prepare_in(repo, repo, env, run_id_text, reader, host, opts).await
508}
509
510pub async fn deploy_rollback_prepare_in(
513 repo: &Path,
514 targets: &Path,
515 env: &str,
516 run_id_text: &str,
517 reader: &dyn TreeReader,
518 host: &dyn ApplyHost,
519 opts: &RollbackPrepareOptions,
520) -> Result<RollbackPrepared, CoreError> {
521 let run_id = RunId::parse(run_id_text)?;
522 let (target_file, target) = load_target(targets, env)?;
523 let checked = check_rollback(
524 env,
525 repo,
526 &target,
527 reader,
528 host,
529 &run_id,
530 opts.discard_live_changes,
531 )
532 .await?;
533 Ok(RollbackPrepared {
534 prose: rollback_prose(run_id.as_str(), &target, &checked),
535 run_id: run_id.as_str().to_string(),
536 record: checked.record,
537 interrupted: checked.interrupted,
538 live_changed: checked.live_changed,
539 target,
540 target_file,
541 })
542}
543
544pub async fn deploy_rollback_execute(
546 prepared: RollbackPrepared,
547 ctx: &RollbackContext<'_>,
548 opts: &RollbackOptions,
549) -> Result<DeployRollbackResult, CoreError> {
550 if prepared.run_id != opts.run_id.as_str() {
551 return Err(CoreError::Internal(
552 "the prepared rollback is for another run than the options name".to_string(),
553 ));
554 }
555 let outcome = run_rollback(ctx, opts).await?;
556 Ok(DeployRollbackResult {
557 env: ctx.env.to_string(),
558 gateway: ctx.target.gateway.clone(),
559 mode: ctx.target.mode.clone(),
560 run_id: outcome.run_id,
561 target_run_id: outcome.target_run_id,
562 run_dir: outcome.run_dir,
563 outcome: outcome.kind,
564 ok: true,
565 restored: outcome.restored,
566 deleted: outcome.deleted,
567 discarded_live_changes: outcome.discarded_live_changes,
568 warnings: outcome.warnings,
569 restart_pending: outcome.restart_pending,
570 restart_pending_new: outcome.restart_pending_new,
571 })
572}
573
574pub async fn deploy_rollback_prepare_for_env(
579 repo: &Path,
580 env: &str,
581 run_id_text: &str,
582 reader: &dyn TreeReader,
583 opts: &RollbackPrepareOptions,
584) -> Result<RollbackPrepared, CoreError> {
585 deploy_rollback_prepare_for_env_in(repo, repo, env, run_id_text, reader, opts).await
586}
587
588pub async fn deploy_rollback_prepare_for_env_in(
591 repo: &Path,
592 targets: &Path,
593 env: &str,
594 run_id_text: &str,
595 reader: &dyn TreeReader,
596 opts: &RollbackPrepareOptions,
597) -> Result<RollbackPrepared, CoreError> {
598 RunId::parse(run_id_text)?;
599 let (target_file, target) = load_target(targets, env)?;
600 let host = apply_host_for_node(
601 targets,
602 &target_file,
603 "nodes.master",
604 &target.nodes.master,
605 &target.gateway,
606 &target.mode,
607 )?;
608 deploy_rollback_prepare_in(repo, targets, env, run_id_text, reader, host.as_ref(), opts).await
609}
610
611#[derive(Debug, Clone, Copy, Default)]
613pub struct RollbackRunOptions {
614 pub discard_live_changes: bool,
616 pub designers: DesignersPolicy,
618}
619
620pub async fn deploy_rollback_run(
627 prepared: RollbackPrepared,
628 env: &str,
629 repo: &Path,
630 reader: &dyn TreeReader,
631 surfaces: &ApplySurfaces<'_>,
632 opts: &RollbackRunOptions,
633 observer: &dyn ApplyObserver,
634) -> Result<DeployRollbackResult, CoreError> {
635 deploy_rollback_run_in(prepared, env, repo, repo, reader, surfaces, opts, observer).await
636}
637
638#[allow(clippy::too_many_arguments)]
641pub async fn deploy_rollback_run_in(
642 prepared: RollbackPrepared,
643 env: &str,
644 repo: &Path,
645 targets: &Path,
646 reader: &dyn TreeReader,
647 surfaces: &ApplySurfaces<'_>,
648 opts: &RollbackRunOptions,
649 observer: &dyn ApplyObserver,
650) -> Result<DeployRollbackResult, CoreError> {
651 let host = apply_host_for_node(
652 targets,
653 &prepared.target_file,
654 "nodes.master",
655 &prepared.target.nodes.master,
656 &prepared.target.gateway,
657 &prepared.target.mode,
658 )?;
659 let run_id = RunId::parse(&prepared.run_id)?;
660 let target = prepared.target.clone();
661 let target_file = prepared.target_file.clone();
662 let ctx = RollbackContext {
663 env,
664 repo,
665 target_file: &target_file,
666 target: &target,
667 reader,
668 host: host.as_ref(),
669 deploy: surfaces.deploy,
670 gateway: surfaces.gateway,
671 backup: surfaces.backup,
672 clock: &SystemClock,
673 observer,
674 };
675 let engine = RollbackOptions {
676 run_id,
677 discard_live_changes: opts.discard_live_changes,
678 designers: opts.designers,
679 hold: HoldModel::MEASURED,
680 poll: scan_poll("config scan after rollback"),
681 };
682 deploy_rollback_execute(prepared, &ctx, &engine).await
683}