1use std::collections::{BTreeMap, BTreeSet};
82use std::path::{Path, PathBuf};
83use std::time::Duration;
84
85use serde::Serialize;
86
87use crate::client::logs::{DEFAULT_LOG_LIMIT, LogQuery};
88use crate::client::mode::normalize_deployment_mode;
89use crate::client::query::ListQuery;
90use crate::client::scan::{ScanKind, ScanStatus};
91use crate::client::{DeployApi, GatewayApi};
92use crate::deploy::apply::journal::{journal_on_box, journal_present};
93use crate::deploy::component_problem;
94use crate::deploy::ledger::{Ledger, LedgerRoot, ledger_location};
95use crate::deploy::manifest::Manifest;
96use crate::deploy::pair::{
97 BackupCheck, BackupCheckId, BackupNode, BackupReport, BackupSignals, SecretCheckMode,
98 SecretNamesReport, TickOutcome, check_secret_names, collect_referenced_secrets, compare_files,
99 declared_secret_providers, effective_files, files_check, files_skipped_check,
100 mode_backup_check, read_back_backup_check, unreachable_check, verify_report,
101};
102use crate::deploy::plan::{BaseTree, PlanEntry, PlanSummary, apply_base, plan, render_diffs};
103use crate::deploy::pull::{PullInput, Skipped, generate_manifest, pull};
104use crate::deploy::render::{RenderInput, RenderOutcome, RenderedResource, render};
105use crate::deploy::target::{NodeTarget, Target};
106use crate::deploy::transport::{CORE_COLLECTION, DeleteSet, READ_MAX_ENTRIES, Scope, TreeReader};
107use crate::deploy::tree::{
108 RenderMarker, SourceTree, read_repo_file, require_repo_dir, walk_capped, write_out,
109};
110use crate::deploy::verify::{self, CheckStatus};
111use crate::error::CoreError;
112
113#[derive(Debug, Clone, PartialEq, Serialize)]
115pub struct DeployRenderResult {
116 pub env: String,
118 pub gateway: String,
120 pub mode: String,
122 pub out: Option<String>,
124 pub files: usize,
126 pub bytes: usize,
128 pub render_sha256: String,
130 pub resources: Vec<RenderedResource>,
132 pub tag_providers: Vec<String>,
134 pub uuids_minted: Vec<String>,
136 pub opaque_files: Vec<String>,
138 pub warnings: Vec<String>,
140}
141
142pub fn load_target(repo: &Path, env: &str) -> Result<(String, Target), CoreError> {
145 if let Some(problem) = component_problem(env) {
146 return Err(CoreError::InvalidInput {
147 reason: format!("--env {env}: {problem}"),
148 });
149 }
150 let target_rel = format!("targets/{env}.yaml");
151 let target_bytes = read_repo_file(
152 repo,
153 &target_rel,
154 Some(&format!("--env {env} selects {target_rel}")),
155 )?;
156 let target = Target::parse(&target_rel, &target_bytes)?;
157 Ok((target_rel, target))
158}
159
160#[derive(Debug, Clone)]
162pub struct Declared {
163 pub env: String,
165 pub target_file: String,
167 pub target: Target,
169 pub manifest: Manifest,
171 pub outcome: RenderOutcome,
173}
174
175pub fn load_declared(repo: &Path, env: &str) -> Result<Declared, CoreError> {
178 load_declared_in(repo, repo, env)
179}
180
181pub fn load_declared_in(repo: &Path, targets: &Path, env: &str) -> Result<Declared, CoreError> {
187 let (target_file, target) = load_target(targets, env)?;
188
189 let gateway_rel = format!("gateways/{}", target.gateway);
190 let gateway_dir = repo.join("gateways").join(&target.gateway);
191 require_repo_dir(repo, "gateways")?;
192 require_repo_dir(repo, &gateway_rel)?;
193
194 let manifest_rel = format!("{gateway_rel}/manifest.yaml");
195 let manifest_bytes = read_repo_file(repo, &manifest_rel, None)?;
196 let manifest = Manifest::parse(&manifest_rel, &manifest_bytes)?;
197
198 let source = SourceTree::read_gateway(&gateway_dir, env)?;
199 let outcome = render(&RenderInput {
200 gateway: &target.gateway,
201 env,
202 manifest: &manifest,
203 target: &target,
204 source: &source,
205 })?;
206 Ok(Declared {
207 env: env.to_string(),
208 target_file,
209 target,
210 manifest,
211 outcome,
212 })
213}
214
215pub fn deploy_render(
218 repo: &Path,
219 env: &str,
220 out: Option<&Path>,
221) -> Result<DeployRenderResult, CoreError> {
222 deploy_render_in(repo, repo, env, out)
223}
224
225pub fn deploy_render_in(
229 repo: &Path,
230 targets: &Path,
231 env: &str,
232 out: Option<&Path>,
233) -> Result<DeployRenderResult, CoreError> {
234 let Declared {
235 target, outcome, ..
236 } = load_declared_in(repo, targets, env)?;
237
238 if let Some(out) = out {
240 let marker = RenderMarker::for_tree(&target.gateway, env, &target.mode, &outcome.tree);
241 write_out(&outcome.tree, out, &marker)?;
242 }
243
244 Ok(DeployRenderResult {
245 env: env.to_string(),
246 gateway: target.gateway.clone(),
247 mode: target.mode.clone(),
248 out: out.map(|p| p.to_string_lossy().into_owned()),
249 files: outcome.tree.files.len(),
250 bytes: outcome.tree.bytes(),
251 render_sha256: outcome.tree.render_sha256(),
252 resources: outcome.resources,
253 tag_providers: outcome.tag_providers,
254 uuids_minted: outcome.uuids_minted,
255 opaque_files: outcome.opaque_files,
256 warnings: outcome.warnings,
257 })
258}
259
260#[derive(Debug, Clone, PartialEq, Serialize)]
262pub struct DeployPlanResult {
263 pub env: String,
265 pub gateway: String,
267 pub mode: String,
269 pub node: String,
271 pub transport: String,
273 pub render_sha256: String,
275 pub plan_sha256: String,
277 pub summary: PlanSummary,
279 pub entries: Vec<PlanEntry>,
281 pub delete_set: DeleteSet,
284 pub base: Option<BaseSummary>,
286 pub warnings: Vec<String>,
288}
289
290#[derive(Debug, Clone, Default)]
292pub struct PlanOptions {
293 pub base: Option<PathBuf>,
296 pub diff: bool,
298}
299
300#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
302pub struct BaseSummary {
303 pub dir: String,
305 pub render_sha256: String,
307 pub files: usize,
309}
310
311pub async fn deploy_plan(
315 declared: &Declared,
316 reader: &dyn TreeReader,
317 opts: &PlanOptions,
318) -> Result<DeployPlanResult, CoreError> {
319 let node = &declared.target.nodes.master;
320 let transport = match (&node.ssh, &node.local) {
321 (Some(_), _) => "ssh",
322 (None, Some(_)) => "local",
323 (None, None) => "none",
324 };
325 let base = match &opts.base {
328 Some(dir) => {
329 let dir = dir.clone();
330 let (gateway, env, mode) = (
331 declared.target.gateway.clone(),
332 declared.env.clone(),
333 declared.target.mode.clone(),
334 );
335 let read =
336 tokio::task::spawn_blocking(move || BaseTree::read(&dir, &gateway, &env, &mode))
337 .await
338 .map_err(|e| CoreError::Internal(format!("--base read task failed: {e}")))?;
339 Some(read?)
340 }
341 None => None,
342 };
343 if let Some(journal) = journal_on_box(
348 reader,
349 node,
350 &declared.target.gateway,
351 &declared.target.mode,
352 )
353 .await?
354 {
355 return Err(journal_present(
356 &journal,
357 &declared.env,
358 node,
359 &declared.target.gateway,
360 &declared.target.mode,
361 ));
362 }
363 let scope = Scope::from_manifest(&declared.manifest, &declared.target.mode);
364 let observed = reader.read(&scope).await?;
365 let gateway = &declared.target.gateway;
366 let mode = &declared.target.mode;
367 let ledger = match reader.read_ledger(gateway, mode).await? {
370 Some(bytes) => Some(Ledger::parse(
371 &ledger_location(LedgerRoot::for_node(node), gateway, mode),
372 &bytes,
373 gateway,
374 mode,
375 )?),
376 None => None,
377 };
378 let mut planned = plan(&declared.outcome.tree, &observed, ledger.as_ref(), &scope);
379 if let Some(base) = &base {
380 apply_base(&mut planned, base);
381 }
382 if opts.diff {
383 render_diffs(&mut planned, &declared.outcome.tree, &observed);
384 }
385
386 let mut warnings = declared.outcome.warnings.clone();
387 warnings.extend(planned.warnings);
388 Ok(DeployPlanResult {
389 env: declared.env.clone(),
390 gateway: declared.target.gateway.clone(),
391 mode: declared.target.mode.clone(),
392 node: "master".to_string(),
393 transport: transport.to_string(),
394 render_sha256: declared.outcome.tree.render_sha256(),
395 plan_sha256: planned.plan_sha256,
396 summary: planned.summary,
397 entries: planned.entries,
398 delete_set: planned.delete_set,
399 base: base.map(|b| BaseSummary {
400 dir: b.dir,
401 render_sha256: b.render_sha256,
402 files: b.files.len(),
403 }),
404 warnings,
405 })
406}
407
408#[derive(Debug, Clone, PartialEq, Serialize)]
410pub struct DeployPullResult {
411 pub env: String,
413 pub gateway: String,
415 pub mode: String,
417 pub node: String,
419 pub transport: String,
421 pub out: String,
423 pub files_written: usize,
426 pub manifest: String,
428 pub render_sha256: String,
431 pub skipped: Vec<Skipped>,
434 pub warnings: Vec<String>,
436}
437
438struct Existing {
442 files: Vec<String>,
443 dirs: BTreeSet<String>,
445}
446
447fn repo_entry_exists(repo: &Path, rel: &str) -> Result<bool, CoreError> {
449 match std::fs::symlink_metadata(repo.join(rel)) {
450 Ok(_) => Ok(true),
451 Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(false),
452 Err(e) => Err(CoreError::deploy_tree_invalid(
453 rel,
454 None,
455 None,
456 format!("cannot inspect ({})", e.kind()),
457 )),
458 }
459}
460
461fn scan_existing(repo: &Path, gateway: &str, env: &str) -> Result<Option<Existing>, CoreError> {
465 let gateway_rel = format!("gateways/{gateway}");
466 for rel in ["gateways", gateway_rel.as_str()] {
467 if !repo_entry_exists(repo, rel)? {
468 return Ok(None);
469 }
470 require_repo_dir(repo, rel)?;
471 }
472 let mut existing = Existing {
473 files: Vec::new(),
474 dirs: BTreeSet::new(),
475 };
476 for sub in [
477 "resources".to_string(),
478 "tags".to_string(),
479 format!("modes/{env}"),
480 ] {
481 if sub.starts_with("modes/") {
482 let modes = format!("{gateway_rel}/modes");
483 if !repo_entry_exists(repo, &modes)? {
484 continue;
485 }
486 require_repo_dir(repo, &modes)?;
487 }
488 let rel = format!("{gateway_rel}/{sub}");
489 if !repo_entry_exists(repo, &rel)? {
490 continue;
491 }
492 require_repo_dir(repo, &rel)?;
493 let mut files: BTreeMap<String, PathBuf> = BTreeMap::new();
494 let mut dirs: BTreeSet<String> = BTreeSet::new();
495 walk_capped(
496 &repo.join(&rel),
497 &sub,
498 &mut files,
499 &mut dirs,
500 READ_MAX_ENTRIES,
501 )
502 .map_err(|why| CoreError::deploy_tree_invalid(&rel, None, None, why))?;
503 existing.dirs.insert(sub.clone());
504 existing.dirs.extend(dirs);
505 existing.files.extend(files.into_keys());
506 }
507 Ok(Some(existing))
508}
509
510fn pull_prose(gateway: &str, env: &str, existing: &Existing) -> String {
512 let shown: Vec<&str> = existing.files.iter().take(5).map(String::as_str).collect();
513 let more = if existing.files.len() > shown.len() {
514 ", ..."
515 } else {
516 ""
517 };
518 format!(
519 "deploy pull would replace gateways/{gateway}/resources, tags and modes/{env} with the live tree of nodes.master ({} existing files, e.g. {}{more})",
520 existing.files.len(),
521 shown.join(", ")
522 )
523}
524
525pub fn deploy_pull_preview(repo: &Path, env: &str) -> Result<Option<String>, CoreError> {
530 deploy_pull_preview_in(repo, repo, env)
531}
532
533pub fn deploy_pull_preview_in(
536 repo: &Path,
537 targets: &Path,
538 env: &str,
539) -> Result<Option<String>, CoreError> {
540 let (_, target) = load_target(targets, env)?;
541 Ok(scan_existing(repo, &target.gateway, env)?
542 .filter(|existing| !existing.files.is_empty())
543 .map(|existing| pull_prose(&target.gateway, env, &existing)))
544}
545
546fn transport_label(node: &NodeTarget) -> &'static str {
547 match (&node.ssh, &node.local) {
548 (Some(_), _) => "ssh",
549 (None, Some(_)) => "local",
550 (None, None) => "none",
551 }
552}
553
554fn live_providers(observed: &crate::deploy::transport::ObservedFiles) -> Vec<String> {
556 let names: BTreeSet<String> = observed
557 .core
558 .keys()
559 .chain(observed.mode.keys())
560 .filter_map(|path| {
561 let parts: Vec<&str> = path.components().collect();
562 (parts.len() >= 4 && parts[0] == "ignition" && parts[1] == "tag-provider")
564 .then(|| parts[2].to_string())
565 })
566 .collect();
567 names.into_iter().collect()
568}
569
570pub async fn deploy_pull(
576 repo: &Path,
577 env: &str,
578 reader: &dyn TreeReader,
579 yes: bool,
580) -> Result<DeployPullResult, CoreError> {
581 deploy_pull_in(repo, repo, env, reader, yes).await
582}
583
584pub async fn deploy_pull_in(
587 repo: &Path,
588 targets: &Path,
589 env: &str,
590 reader: &dyn TreeReader,
591 yes: bool,
592) -> Result<DeployPullResult, CoreError> {
593 let (_, target) = load_target(targets, env)?;
594 let gateway = target.gateway.clone();
595 let existing = scan_existing(repo, &gateway, env)?;
596 if let Some(existing) = existing.as_ref().filter(|e| !e.files.is_empty())
597 && !yes
598 {
599 return Err(CoreError::ConfirmationRequired {
600 operation: pull_prose(&gateway, env, existing),
601 });
602 }
603
604 let manifest_rel = format!("gateways/{gateway}/manifest.yaml");
607 let mut left_out: Vec<Skipped> = Vec::new();
608 let (manifest, generated) = if repo_entry_exists(repo, &manifest_rel)? {
609 let bytes = read_repo_file(repo, &manifest_rel, None)?;
610 (Manifest::parse(&manifest_rel, &bytes)?, None)
611 } else {
612 let mut live_types = reader.list_types(CORE_COLLECTION).await?;
613 live_types.extend(reader.list_types(&target.mode).await?);
614 let discovery = Scope {
615 mode: target.mode.clone(),
616 types: vec![("ignition".to_string(), "tag-provider".to_string())],
617 tag_providers: Vec::new(),
618 };
619 let providers = live_providers(&reader.read(&discovery).await?);
620 let (yaml, skipped) = generate_manifest(&live_types, &providers);
621 left_out = skipped;
622 let manifest = Manifest::parse(&manifest_rel, yaml.as_bytes())?;
623 (manifest, Some(yaml))
624 };
625
626 let scope = Scope::from_manifest(&manifest, &target.mode);
627 let observed = reader.read(&scope).await?;
628 let outcome = pull(&PullInput {
629 gateway: &gateway,
630 env,
631 target: &target,
632 manifest: &manifest,
633 observed: &observed,
634 })?;
635
636 write_pulled(
637 repo,
638 &gateway,
639 existing.as_ref(),
640 &outcome.source,
641 generated.as_deref(),
642 )?;
643
644 left_out.extend(outcome.skipped);
645 left_out.sort_by(|a, b| a.path.as_bytes().cmp(b.path.as_bytes()));
646 let mut warnings = outcome.rendered.warnings.clone();
647 warnings.extend(observed.local_units.iter().map(|unit| {
648 format!(
649 "{unit} also exists in the `local` collection, which the gateway loads over the mode collection; pull did not read the `local` copy"
650 )
651 }));
652 Ok(DeployPullResult {
653 env: env.to_string(),
654 gateway: gateway.clone(),
655 mode: target.mode.clone(),
656 node: "master".to_string(),
657 transport: transport_label(&target.nodes.master).to_string(),
658 out: format!("gateways/{gateway}"),
659 files_written: outcome.source.files.len(),
660 manifest: if generated.is_some() {
661 "generated"
662 } else {
663 "kept"
664 }
665 .to_string(),
666 render_sha256: outcome.rendered.tree.render_sha256(),
667 skipped: left_out,
668 warnings,
669 })
670}
671
672fn safe_component(name: &str) -> bool {
675 !name.is_empty()
676 && name != "."
677 && name != ".."
678 && !name.contains(['/', '\\', '\0'])
679 && !name.chars().any(char::is_control)
680}
681
682fn validate_pulled(gateway: &str, source: &SourceTree) -> Result<(), CoreError> {
688 let refuse = |path: &str, why: String| {
689 CoreError::deploy_tree_invalid(format!("gateways/{gateway}/{path}"), None, None, why)
690 };
691 let mut folded: BTreeMap<String, &str> = BTreeMap::new();
692 let mut dir_folds: BTreeSet<String> = BTreeSet::new();
693 for path in source.files.keys() {
694 let parts: Vec<&str> = path.components().collect();
695 if parts.iter().any(|c| !safe_component(c)) {
696 return Err(refuse(
697 path.as_str(),
698 "has an unsafe path component; refused before anything was changed".to_string(),
699 ));
700 }
701 let fold = path.as_str().to_lowercase();
702 if let Some(other) = folded.insert(fold.clone(), path.as_str()) {
703 return Err(refuse(
704 path.as_str(),
705 format!(
706 "collides with {other} when case is folded; a case-insensitive filesystem keeps only one. Refused before anything was changed"
707 ),
708 ));
709 }
710 for end in fold.match_indices('/').map(|(i, _)| i) {
711 dir_folds.insert(fold[..end].to_string());
712 }
713 }
714 for (fold, path) in &folded {
715 if dir_folds.contains(fold) {
716 return Err(refuse(
717 path,
718 "is both a file and a directory in the pulled tree; refused before anything was changed"
719 .to_string(),
720 ));
721 }
722 }
723 Ok(())
724}
725
726fn write_pulled(
730 repo: &Path,
731 gateway: &str,
732 existing: Option<&Existing>,
733 source: &SourceTree,
734 manifest: Option<&str>,
735) -> Result<(), CoreError> {
736 validate_pulled(gateway, source)?;
737 write_pulled_fs(repo, gateway, existing, source, manifest)
738}
739
740#[cfg(unix)]
744fn write_pulled_fs(
745 repo: &Path,
746 gateway: &str,
747 existing: Option<&Existing>,
748 source: &SourceTree,
749 manifest: Option<&str>,
750) -> Result<(), CoreError> {
751 use crate::deploy::tree::dirfd::{self, FdError, Kind};
752 use std::io::Write;
753
754 let fail = |what: &str, e: FdError| match e {
755 FdError::Changed => CoreError::deploy_tree_invalid(
756 what,
757 None,
758 None,
759 "changed while it was written (a symlink or a non-directory appeared); refused, nothing outside the gateway directory was touched",
760 ),
761 FdError::Io(io) => CoreError::Internal(format!("cannot write {what}: {io}")),
762 other => CoreError::Internal(format!("cannot write {what}: {other:?}")),
763 };
764
765 let repo_real = std::fs::canonicalize(repo).map_err(|e| {
769 CoreError::Internal(format!("cannot resolve the repo {}: {e}", repo.display()))
770 })?;
771 let root = dirfd::open_root(&repo_real).map_err(|e| fail("the repo", e))?;
772 let gateways = dirfd::ensure_dir(&root, "gateways").map_err(|e| fail("gateways", e))?;
773 let gw_label = format!("gateways/{gateway}");
774 let gw = dirfd::ensure_dir(&gateways, gateway).map_err(|e| fail(&gw_label, e))?;
775
776 if let Some(existing) = existing {
777 for file in &existing.files {
778 let parts: Vec<&str> = file.split('/').collect();
779 let (name, parents) = parts.split_last().ok_or_else(|| {
780 CoreError::Internal("an existing file has an empty path".to_string())
781 })?;
782 let label = format!("{gw_label}/{file}");
783 let dir = dirfd::walk(&gw, parents).map_err(|e| fail(&label, e))?;
784 match dirfd::kind_at(&dir, name).map_err(|e| fail(&label, e))? {
785 Kind::File => dirfd::unlink_at(&dir, name).map_err(|e| fail(&label, e))?,
786 Kind::Missing => {}
787 _ => return Err(fail(&label, FdError::Changed)),
788 }
789 }
790 for dir_path in existing.dirs.iter().rev() {
792 let parts: Vec<&str> = dir_path.split('/').collect();
793 let Some((name, parents)) = parts.split_last() else {
794 continue;
795 };
796 let label = format!("{gw_label}/{dir_path}");
797 let parent = dirfd::walk(&gw, parents).map_err(|e| fail(&label, e))?;
798 match dirfd::rmdir_at(&parent, name) {
799 Ok(()) | Err(FdError::Missing) => {}
800 Err(FdError::NotEmpty) => {
801 return Err(CoreError::deploy_tree_invalid(
802 label,
803 None,
804 None,
805 "is not empty after the pull removed its files (something wrote into it); nothing more was changed",
806 ));
807 }
808 Err(e) => return Err(fail(&label, e)),
809 }
810 }
811 }
812
813 if let Some(text) = manifest {
814 let label = format!("{gw_label}/manifest.yaml");
815 let mut file = dirfd::create_new(&gw, "manifest.yaml").map_err(|e| fail(&label, e))?;
816 file.write_all(text.as_bytes())
817 .map_err(|e| CoreError::Internal(format!("cannot write {label}: {e}")))?;
818 }
819 for (path, bytes) in &source.files {
820 let label = format!("{gw_label}/{path}");
821 let parts: Vec<&str> = path.components().collect();
822 if parts.iter().any(|c| !safe_component(c)) {
823 return Err(CoreError::deploy_tree_invalid(
824 label,
825 None,
826 None,
827 "has an unsafe path component; refused",
828 ));
829 }
830 let (name, parents) = parts
831 .split_last()
832 .ok_or_else(|| CoreError::Internal("a pulled file has an empty path".to_string()))?;
833 let dir = dirfd::ensure_walk(&gw, parents).map_err(|e| fail(&label, e))?;
834 let mut file = dirfd::create_new(&dir, name).map_err(|e| fail(&label, e))?;
835 file.write_all(bytes)
836 .map_err(|e| CoreError::Internal(format!("cannot write {label}: {e}")))?;
837 }
838 Ok(())
839}
840
841#[cfg(not(unix))]
845fn write_pulled_fs(
846 repo: &Path,
847 gateway: &str,
848 existing: Option<&Existing>,
849 source: &SourceTree,
850 manifest: Option<&str>,
851) -> Result<(), CoreError> {
852 use std::io::Write;
853
854 let io = |what: &Path, e: std::io::Error| {
855 CoreError::Internal(format!("cannot write {}: {e}", what.display()))
856 };
857 let gw = repo.join("gateways").join(gateway);
858 std::fs::create_dir_all(&gw).map_err(|e| io(&gw, e))?;
859 if let Some(existing) = existing {
860 for file in &existing.files {
861 let path = file.split('/').fold(gw.clone(), |acc, c| acc.join(c));
862 std::fs::remove_file(&path).map_err(|e| io(&path, e))?;
863 }
864 for dir_path in existing.dirs.iter().rev() {
865 let path = dir_path.split('/').fold(gw.clone(), |acc, c| acc.join(c));
866 std::fs::remove_dir(&path).map_err(|e| io(&path, e))?;
867 }
868 }
869 let create = |path: &Path, bytes: &[u8]| -> Result<(), CoreError> {
870 if let Some(parent) = path.parent() {
871 std::fs::create_dir_all(parent).map_err(|e| io(parent, e))?;
872 }
873 let mut file = std::fs::OpenOptions::new()
874 .write(true)
875 .create_new(true)
876 .open(path)
877 .map_err(|e| io(path, e))?;
878 file.write_all(bytes).map_err(|e| io(path, e))
879 };
880 if let Some(text) = manifest {
881 create(&gw.join("manifest.yaml"), text.as_bytes())?;
882 }
883 for (path, bytes) in &source.files {
884 let parts: Vec<&str> = path.components().collect();
885 if parts.iter().any(|c| !safe_component(c)) {
886 return Err(CoreError::deploy_tree_invalid(
887 format!("gateways/{gateway}/{path}"),
888 None,
889 None,
890 "has an unsafe path component; refused",
891 ));
892 }
893 let dest = parts.iter().fold(gw.clone(), |acc, c| acc.join(c));
894 create(&dest, bytes)?;
895 }
896 Ok(())
897}
898
899#[derive(Debug, Clone)]
901pub struct VerifyOptions {
902 pub probe_deadline: Duration,
904 pub probe_interval: Duration,
906 pub log_settle: Duration,
909}
910
911impl Default for VerifyOptions {
912 fn default() -> Self {
913 Self {
914 probe_deadline: verify::PROBE_DEADLINE,
915 probe_interval: verify::PROBE_INTERVAL,
916 log_settle: Duration::from_millis(verify::LOG_SETTLE_MS.unsigned_abs()),
917 }
918 }
919}
920
921#[derive(Debug, Clone, PartialEq, Serialize)]
923pub struct DeployVerifyResult {
924 pub env: String,
926 pub gateway: String,
928 pub mode: String,
930 pub node: String,
933 pub project: String,
935 pub ok: bool,
937 pub slug: Option<&'static str>,
939 pub checks: Vec<verify::Check>,
941 pub resources_checked: usize,
943 pub warnings: Vec<String>,
945 #[serde(skip_serializing_if = "Option::is_none")]
948 pub backup: Option<BackupReport>,
949 #[serde(skip_serializing_if = "Option::is_none")]
952 pub secrets: Option<SecretNamesReport>,
953}
954
955impl DeployVerifyResult {
956 pub fn failure_exit_code(&self) -> Option<u8> {
961 (!self.ok).then_some(6)
962 }
963}
964
965const LOG_PAGES_MAX: usize = 5;
967
968pub async fn deploy_verify(
983 declared: &Declared,
984 project: &str,
985 deploy: &dyn DeployApi,
986 gateway: &dyn GatewayApi,
987 reader: &dyn TreeReader,
988 opts: &VerifyOptions,
989) -> Result<DeployVerifyResult, CoreError> {
990 let expected =
991 verify::Expected::from_render(&declared.target.mode, &declared.outcome, &declared.manifest);
992
993 let scan = deploy.scan_status(ScanKind::Config).await?;
994 let scan_read_at = std::time::Instant::now();
995 let mode = normalize_deployment_mode(deploy.deployment_mode().await?);
996 let read_back = gather_read_back(deploy, &expected).await?;
997
998 let scope = Scope::from_manifest(&declared.manifest, &declared.target.mode);
1001 let live = reader.read(&scope).await?;
1002 let planned = plan(&declared.outcome.tree, &live, None, &scope);
1003 let files = planned
1004 .entries
1005 .iter()
1006 .filter(|entry| expected.tag_units.contains(&entry.key))
1007 .map(|entry| (entry.key.clone(), entry.status))
1008 .collect();
1009
1010 let logs = if scan.scan_active {
1011 Ok(Vec::new())
1013 } else {
1014 settle_logs(gateway, &scan, scan_read_at, opts.log_settle).await?;
1015 read_logs(gateway, verify::log_window_start(&scan)).await?
1016 };
1017 let probes = if expected.probes.is_empty() {
1018 None
1019 } else {
1020 Some(poll_probes(gateway, project, &expected.probes, opts).await?)
1021 };
1022
1023 let observed = verify::ObservedState {
1024 scan,
1025 mode,
1026 read_back,
1027 files,
1028 logs,
1029 probes,
1030 };
1031 let report = verify::evaluate(&expected, &observed);
1032 Ok(DeployVerifyResult {
1033 env: declared.env.clone(),
1034 gateway: declared.target.gateway.clone(),
1035 mode: declared.target.mode.clone(),
1036 node: "master".to_string(),
1037 project: project.to_string(),
1038 ok: report.ok,
1039 slug: report.slug,
1040 checks: report.checks,
1041 resources_checked: report.resources_checked,
1042 warnings: report.warnings,
1043 backup: None,
1044 secrets: None,
1045 })
1046}
1047
1048#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1050pub enum VerifyNode {
1051 Master,
1054 Backup,
1056 Both,
1058}
1059
1060impl VerifyNode {
1061 pub fn as_str(self) -> &'static str {
1063 match self {
1064 VerifyNode::Master => "master",
1065 VerifyNode::Backup => "backup",
1066 VerifyNode::Both => "both",
1067 }
1068 }
1069}
1070
1071pub type VerifyMasterSurfaces<'a> = (&'a dyn DeployApi, &'a dyn GatewayApi);
1073
1074pub async fn deploy_verify_node(
1088 declared: &Declared,
1089 project: &str,
1090 node: VerifyNode,
1091 master: Option<VerifyMasterSurfaces<'_>>,
1092 master_reader: &dyn TreeReader,
1093 backup: Option<&BackupNode<'_>>,
1094 opts: &VerifyOptions,
1095) -> Result<DeployVerifyResult, CoreError> {
1096 let missing = |what: &str| CoreError::InvalidInput {
1097 reason: format!("deploy verify --node {} needs the {what}", node.as_str()),
1098 };
1099 let master_result = if node == VerifyNode::Backup {
1100 None
1101 } else {
1102 let (deploy, gateway) = master.ok_or_else(|| missing("master surfaces"))?;
1103 Some(deploy_verify(declared, project, deploy, gateway, master_reader, opts).await?)
1104 };
1105 if node == VerifyNode::Master {
1106 return master_result.ok_or_else(|| missing("master surfaces"));
1107 }
1108 let backup = backup.ok_or_else(|| missing("backup node"))?;
1109 let expected =
1110 verify::Expected::from_render(&declared.target.mode, &declared.outcome, &declared.manifest);
1111 let mut backup_warnings = Vec::new();
1112 let (tick, compared) = verify_backup_tick(
1113 declared,
1114 &expected,
1115 master_reader,
1116 backup,
1117 &mut backup_warnings,
1118 )
1119 .await;
1120 let report = verify_report(&tick);
1121
1122 let mut result = match master_result {
1123 Some(mut result) => {
1127 result.warnings.extend(backup_warnings);
1128 result
1129 }
1130 None => DeployVerifyResult {
1131 env: declared.env.clone(),
1132 gateway: declared.target.gateway.clone(),
1133 mode: declared.target.mode.clone(),
1134 node: String::new(),
1135 project: project.to_string(),
1136 ok: true,
1137 slug: None,
1138 checks: Vec::new(),
1139 resources_checked: compared,
1140 warnings: backup_warnings,
1141 backup: None,
1142 secrets: None,
1143 },
1144 };
1145 let secrets = verify_secret_names(
1148 declared,
1149 master
1150 .map(|(deploy, _)| deploy)
1151 .filter(|_| node != VerifyNode::Backup),
1152 Some(backup),
1153 )
1154 .await?;
1155 let secrets_ok = secrets.nodes.iter().all(|n| n.status != CheckStatus::Fail);
1156 result.node = node.as_str().to_string();
1157 result.ok = result.ok && report.converged && secrets_ok;
1158 result.slug = (!result.ok).then_some(verify::VERIFY_FAILED_SLUG);
1159 result.backup = Some(report);
1160 result.secrets = Some(secrets);
1161 Ok(result)
1162}
1163
1164async fn verify_secret_names(
1171 declared: &Declared,
1172 master: Option<&dyn DeployApi>,
1173 backup: Option<&BackupNode<'_>>,
1174) -> Result<SecretNamesReport, CoreError> {
1175 let referenced = collect_referenced_secrets(&declared.outcome.tree);
1176 let declared_providers = declared_secret_providers(&declared.outcome.tree);
1177 let providers: BTreeSet<&str> = referenced.iter().map(|(p, _)| p.as_str()).collect();
1178 let mut nodes = Vec::with_capacity(2);
1179 if let Some(master) = master {
1180 let mut listings = BTreeMap::new();
1181 for provider in &providers {
1182 listings.insert(
1183 (*provider).to_string(),
1184 master.secret_names(provider).await?,
1185 );
1186 }
1187 nodes.push(check_secret_names(
1188 "master",
1189 &referenced,
1190 &listings,
1191 &declared_providers,
1192 SecretCheckMode::Verify,
1193 ));
1194 }
1195 if let Some(backup) = backup {
1196 let mut listings = BTreeMap::new();
1197 for provider in &providers {
1198 listings.insert(
1199 (*provider).to_string(),
1200 backup.secret_names(provider).await?,
1201 );
1202 }
1203 nodes.push(check_secret_names(
1204 "backup",
1205 &referenced,
1206 &listings,
1207 &declared_providers,
1208 SecretCheckMode::Verify,
1209 ));
1210 }
1211 Ok(SecretNamesReport::ordered(nodes))
1212}
1213
1214async fn verify_backup_tick(
1221 declared: &Declared,
1222 expected: &verify::Expected,
1223 master_reader: &dyn TreeReader,
1224 backup: &BackupNode<'_>,
1225 warnings: &mut Vec<String>,
1226) -> (TickOutcome, usize) {
1227 let signals = match backup.status().await {
1228 Ok(status) => Ok(BackupSignals::from_status(None, &status)),
1229 Err(error) => Err(error.code().to_string()),
1230 };
1231 let mode = match backup.deployment_mode().await {
1232 Ok(mode) => mode_backup_check(expected, mode.as_deref()),
1233 Err(error) => unreachable_check(BackupCheckId::Mode, error.code()),
1234 };
1235 let mut compared = 0;
1236 let read_back = match backup.read_back(expected).await {
1237 Ok(read_back) => {
1238 let (check, count) = read_back_backup_check(expected, &read_back, warnings);
1239 compared = count;
1240 check
1241 }
1242 Err(error) => unreachable_check(BackupCheckId::ReadBack, error.code()),
1243 };
1244 let (files, files_checked) = verify_backup_files(declared, master_reader, backup).await;
1245 (
1246 TickOutcome {
1247 signals,
1248 exhausted: false,
1249 content: Some(vec![mode, read_back, files]),
1250 files_checked,
1251 stable_ticks: 0,
1252 },
1253 compared,
1254 )
1255}
1256
1257async fn verify_backup_files(
1260 declared: &Declared,
1261 master_reader: &dyn TreeReader,
1262 backup: &BackupNode<'_>,
1263) -> (BackupCheck, usize) {
1264 let scope = Scope::from_manifest(&declared.manifest, &declared.target.mode);
1265 let backup_files = match backup.read_files(&scope).await {
1266 None => return (files_skipped_check(), 0),
1268 Some(Ok(observed)) => effective_files(observed),
1269 Some(Err(error)) => {
1270 return (
1271 unreachable_check(BackupCheckId::Files, &format!("backup {}", error.code())),
1272 0,
1273 );
1274 }
1275 };
1276 let master_files = match master_reader.read(&scope).await {
1277 Ok(observed) => effective_files(observed),
1278 Err(error) => {
1279 return (
1280 unreachable_check(BackupCheckId::Files, &format!("master {}", error.code())),
1281 0,
1282 );
1283 }
1284 };
1285 let present: BTreeSet<_> = declared.outcome.tree.files.keys().cloned().collect();
1286 let comparison = compare_files(&master_files, &backup_files, &present, &BTreeSet::new());
1287 (files_check(&comparison), comparison.checked)
1288}
1289
1290pub(crate) async fn gather_read_back(
1295 deploy: &dyn DeployApi,
1296 expected: &verify::Expected,
1297) -> Result<BTreeMap<String, verify::Remote>, CoreError> {
1298 let classify = |record: crate::client::config_resources::ResourceRecord| {
1299 if record.collection.as_deref() == Some(verify::SYSTEM_COLLECTION) {
1300 verify::Remote::System
1301 } else {
1302 verify::Remote::Config(record.config)
1303 }
1304 };
1305 let mut lists: BTreeMap<(String, String), Option<Vec<_>>> = BTreeMap::new();
1306 let mut out = BTreeMap::new();
1307 for res in &expected.resources {
1308 let remote = match &res.name {
1309 None => match deploy.resource_singleton(&res.module, &res.ty, None).await {
1310 Ok(record) => classify(record),
1311 Err(CoreError::NotFound { .. }) => verify::Remote::Missing,
1312 Err(e) => return Err(e),
1313 },
1314 Some(name) => {
1315 let type_key = (res.module.clone(), res.ty.clone());
1316 if !lists.contains_key(&type_key) {
1317 let items = match deploy
1318 .resource_list(&res.module, &res.ty, &ListQuery::default())
1319 .await
1320 {
1321 Ok(envelope) => Some(envelope.items),
1322 Err(CoreError::NotFound { .. }) => None,
1323 Err(e) => return Err(e),
1324 };
1325 lists.insert(type_key.clone(), items);
1326 }
1327 lists
1328 .get(&type_key)
1329 .and_then(Option::as_ref)
1330 .and_then(|items| items.iter().find(|r| r.name.as_deref() == Some(name)))
1331 .map_or(verify::Remote::Missing, |r| classify(r.clone()))
1332 }
1333 };
1334 out.insert(res.key.clone(), remote);
1335 }
1336 Ok(out)
1337}
1338
1339async fn settle_logs(
1343 gateway: &dyn GatewayApi,
1344 scan: &ScanStatus,
1345 scan_read_at: std::time::Instant,
1346 settle: Duration,
1347) -> Result<(), CoreError> {
1348 if settle.is_zero() {
1349 return Ok(());
1350 }
1351 let settle_ms = i64::try_from(settle.as_millis()).unwrap_or(i64::MAX);
1352 let probe = LogQuery {
1353 start_time: Some(scan.last_scan_timestamp.saturating_add(settle_ms)),
1354 limit: 1,
1355 ..LogQuery::default()
1356 };
1357 match gateway.logs(&probe).await {
1358 Ok(envelope) if !envelope.items.is_empty() => return Ok(()),
1359 Err(e @ CoreError::Network { .. }) => return Err(e),
1360 Ok(_) | Err(_) => {}
1362 }
1363 tokio::time::sleep(settle.saturating_sub(scan_read_at.elapsed())).await;
1364 Ok(())
1365}
1366
1367async fn read_logs(
1371 gateway: &dyn GatewayApi,
1372 start_time: i64,
1373) -> Result<Result<Vec<verify::LogLine>, String>, CoreError> {
1374 let mut lines = Vec::new();
1375 let mut offset = 0i64;
1376 for _ in 0..LOG_PAGES_MAX {
1377 let query = LogQuery {
1378 start_time: Some(start_time),
1379 min_level: Some("WARN".to_string()),
1380 limit: DEFAULT_LOG_LIMIT,
1381 offset,
1382 ..LogQuery::default()
1383 };
1384 match gateway.logs(&query).await {
1385 Ok(envelope) => {
1386 let count = i64::try_from(envelope.items.len()).unwrap_or(i64::MAX);
1387 lines.extend(envelope.items.into_iter().map(|e| verify::LogLine {
1388 timestamp: e.timestamp,
1389 logger: e.logger_name,
1390 message: e.message,
1391 }));
1392 if count < DEFAULT_LOG_LIMIT {
1393 return Ok(Ok(lines));
1394 }
1395 offset += count;
1396 }
1397 Err(e @ CoreError::Network { .. }) => return Err(e),
1398 Err(e) => return Ok(Err(e.code().to_string())),
1399 }
1400 }
1401 Ok(Err("logs_truncated".to_string()))
1402}
1403
1404async fn poll_probes(
1409 gateway: &dyn GatewayApi,
1410 project: &str,
1411 paths: &[String],
1412 opts: &VerifyOptions,
1413) -> Result<Result<Vec<verify::ProbeReading>, String>, CoreError> {
1414 let interval = opts.probe_interval.max(Duration::from_millis(1));
1415 let started = tokio::time::Instant::now();
1416 loop {
1417 let readings: Vec<verify::ProbeReading> =
1418 match crate::actions::tags::tags_read(gateway, project, paths).await {
1419 Ok(result) => result
1420 .results
1421 .into_iter()
1422 .enumerate()
1423 .map(|(i, row)| verify::ProbeReading {
1424 path: if row.path.is_empty() {
1426 paths.get(i).cloned().unwrap_or_default()
1427 } else {
1428 row.path
1429 },
1430 quality: row.quality,
1431 })
1432 .collect(),
1433 Err(e @ CoreError::Network { .. }) => return Err(e),
1434 Err(e) => return Ok(Err(e.code().to_string())),
1435 };
1436 let all_good = paths.iter().all(|path| {
1437 readings
1438 .iter()
1439 .any(|r| &r.path == path && r.quality.starts_with("Good"))
1440 });
1441 if all_good || started.elapsed() + interval > opts.probe_deadline {
1442 return Ok(Ok(readings));
1443 }
1444 tokio::time::sleep(interval).await;
1445 }
1446}
1447
1448#[cfg(test)]
1449mod tests {
1450 use super::*;
1451
1452 fn repo_with_existing() -> (tempfile::TempDir, Existing) {
1454 let repo = tempfile::TempDir::new().expect("repo");
1455 let dir = repo.path().join("gateways/g/resources/ignition/x");
1456 std::fs::create_dir_all(&dir).expect("dirs");
1457 std::fs::write(dir.join("keep.json"), b"{\"keep\": true}").expect("file");
1458 let existing = Existing {
1459 files: vec!["resources/ignition/x/keep.json".to_string()],
1460 dirs: ["resources", "resources/ignition", "resources/ignition/x"]
1461 .iter()
1462 .map(|d| d.to_string())
1463 .collect(),
1464 };
1465 (repo, existing)
1466 }
1467
1468 fn snapshot(root: &Path) -> BTreeMap<PathBuf, Vec<u8>> {
1469 fn walk(dir: &Path, out: &mut BTreeMap<PathBuf, Vec<u8>>) {
1470 for entry in std::fs::read_dir(dir).expect("read_dir") {
1471 let path = entry.expect("entry").path();
1472 if path.is_dir() {
1473 out.insert(path.clone(), Vec::new());
1474 walk(&path, out);
1475 } else {
1476 out.insert(path.clone(), std::fs::read(&path).expect("read"));
1477 }
1478 }
1479 }
1480 let mut out = BTreeMap::new();
1481 walk(root, &mut out);
1482 out
1483 }
1484
1485 fn assert_refused_untouched(files: &[(&str, &str)], needle: &str) {
1488 let (repo, existing) = repo_with_existing();
1489 let before = snapshot(repo.path());
1490 let source = SourceTree::from_files(files.iter().copied()).expect("source");
1491 let err =
1492 write_pulled(repo.path(), "g", Some(&existing), &source, None).expect_err("refused");
1493 assert_eq!(err.code(), "deploy_tree_invalid", "{err}");
1494 assert!(err.to_string().contains(needle), "{needle}: {err}");
1495 assert_eq!(
1496 snapshot(repo.path()),
1497 before,
1498 "a refused pull must not delete the existing tree"
1499 );
1500 }
1501
1502 #[test]
1503 fn unsafe_component_is_refused_before_any_delete() {
1504 assert_refused_untouched(
1505 &[
1506 ("resources/ignition/x/fine.json", "{}"),
1507 ("resources/ignition/x/bad\nname.json", "{}"),
1508 ],
1509 "unsafe path component",
1510 );
1511 }
1512
1513 #[test]
1514 fn case_fold_collision_is_refused_before_any_delete() {
1515 assert_refused_untouched(
1516 &[
1517 ("resources/ignition/x/Foo.json", "{}"),
1518 ("resources/ignition/x/foo.json", "{}"),
1519 ],
1520 "case is folded",
1521 );
1522 }
1523
1524 #[test]
1525 fn file_and_directory_clash_is_refused_before_any_delete() {
1526 assert_refused_untouched(
1527 &[
1528 ("resources/ignition/x", "{}"),
1529 ("resources/ignition/x/a.json", "{}"),
1530 ],
1531 "both a file and a directory",
1532 );
1533 }
1534
1535 #[test]
1536 fn a_valid_tree_replaces_the_existing_one() {
1537 let (repo, existing) = repo_with_existing();
1538 let source =
1539 SourceTree::from_files([("resources/ignition/x/new.json", "{}")]).expect("source");
1540 write_pulled(repo.path(), "g", Some(&existing), &source, None).expect("written");
1541 let dir = repo.path().join("gateways/g/resources/ignition/x");
1542 assert!(!dir.join("keep.json").exists());
1543 assert!(dir.join("new.json").is_file());
1544 }
1545}