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::exclude;
95use crate::deploy::layout::{RelPath, collection_path};
96use crate::deploy::ledger::{Ledger, LedgerRoot, ledger_location};
97use crate::deploy::manifest::Manifest;
98use crate::deploy::pair::{
99 BackupCheck, BackupCheckId, BackupNode, BackupReport, BackupSignals, SecretCheckMode,
100 SecretNamesReport, TickOutcome, check_secret_names, collect_referenced_secrets, compare_files,
101 declared_secret_providers, effective_files, exempt_unreadable, files_check,
102 files_skipped_check, mode_backup_check, read_back_backup_check, unreachable_check,
103 unreadable_units, unreadable_warnings, verify_report,
104};
105use crate::deploy::plan::{
106 BaseTree, PlanEntry, PlanSummary, UnreadableUnit, apply_base, plan, render_diffs,
107};
108use crate::deploy::pull::{PullInput, REASON_PATH_TOO_LONG, Skipped, generate_manifest, pull};
109use crate::deploy::render::{RenderInput, RenderOutcome, RenderedResource, render};
110use crate::deploy::target::{NodeTarget, Target};
111use crate::deploy::transport::{CORE_COLLECTION, DeleteSet, READ_MAX_ENTRIES, Scope, TreeReader};
112use crate::deploy::tree::{
113 RenderMarker, SourceTree, read_repo_file, require_repo_dir, walk_capped, write_out,
114};
115use crate::deploy::verify::{self, CheckStatus};
116use crate::error::CoreError;
117
118#[derive(Debug, Clone, PartialEq, Serialize)]
120pub struct DeployRenderResult {
121 pub env: String,
123 pub gateway: String,
125 pub mode: String,
127 pub out: Option<String>,
129 pub files: usize,
131 pub bytes: usize,
133 pub render_sha256: String,
135 pub resources: Vec<RenderedResource>,
137 pub tag_providers: Vec<String>,
139 pub uuids_minted: Vec<String>,
141 pub opaque_files: Vec<String>,
143 pub warnings: Vec<String>,
145}
146
147pub fn load_target(repo: &Path, env: &str) -> Result<(String, Target), CoreError> {
150 if let Some(problem) = component_problem(env) {
151 return Err(CoreError::InvalidInput {
152 reason: format!("--env {env}: {problem}"),
153 });
154 }
155 let target_rel = format!("targets/{env}.yaml");
156 let target_bytes = read_repo_file(
157 repo,
158 &target_rel,
159 Some(&format!("--env {env} selects {target_rel}")),
160 )?;
161 let target = Target::parse(&target_rel, &target_bytes)?;
162 Ok((target_rel, target))
163}
164
165#[derive(Debug, Clone)]
167pub struct Declared {
168 pub env: String,
170 pub target_file: String,
172 pub target: Target,
174 pub manifest: Manifest,
176 pub outcome: RenderOutcome,
178}
179
180impl Declared {
181 pub fn excluded_tag_providers(&self) -> Vec<String> {
185 exclude::effective(&self.target, Some(&self.manifest))
186 }
187}
188
189pub fn load_declared(repo: &Path, env: &str) -> Result<Declared, CoreError> {
192 load_declared_in(repo, repo, env)
193}
194
195pub fn load_declared_in(repo: &Path, targets: &Path, env: &str) -> Result<Declared, CoreError> {
201 let (target_file, target) = load_target(targets, env)?;
202
203 let gateway_rel = format!("gateways/{}", target.gateway);
204 let gateway_dir = repo.join("gateways").join(&target.gateway);
205 require_repo_dir(repo, "gateways")?;
206 require_repo_dir(repo, &gateway_rel)?;
207
208 let manifest_rel = format!("{gateway_rel}/manifest.yaml");
209 let manifest_bytes = read_repo_file(repo, &manifest_rel, None)?;
210 let manifest = Manifest::parse(&manifest_rel, &manifest_bytes)?;
211
212 let source = SourceTree::read_gateway(&gateway_dir, env)?;
213 let outcome = render(&RenderInput {
214 gateway: &target.gateway,
215 env,
216 manifest: &manifest,
217 target: &target,
218 source: &source,
219 })?;
220 Ok(Declared {
221 env: env.to_string(),
222 target_file,
223 target,
224 manifest,
225 outcome,
226 })
227}
228
229pub fn deploy_render(
232 repo: &Path,
233 env: &str,
234 out: Option<&Path>,
235) -> Result<DeployRenderResult, CoreError> {
236 deploy_render_in(repo, repo, env, out)
237}
238
239pub fn deploy_render_in(
243 repo: &Path,
244 targets: &Path,
245 env: &str,
246 out: Option<&Path>,
247) -> Result<DeployRenderResult, CoreError> {
248 let Declared {
249 target, outcome, ..
250 } = load_declared_in(repo, targets, env)?;
251
252 if let Some(out) = out {
254 let marker = RenderMarker::for_tree(&target.gateway, env, &target.mode, &outcome.tree);
255 write_out(&outcome.tree, out, &marker)?;
256 }
257
258 Ok(DeployRenderResult {
259 env: env.to_string(),
260 gateway: target.gateway.clone(),
261 mode: target.mode.clone(),
262 out: out.map(|p| p.to_string_lossy().into_owned()),
263 files: outcome.tree.files.len(),
264 bytes: outcome.tree.bytes(),
265 render_sha256: outcome.tree.render_sha256(),
266 resources: outcome.resources,
267 tag_providers: outcome.tag_providers,
268 uuids_minted: outcome.uuids_minted,
269 opaque_files: outcome.opaque_files,
270 warnings: outcome.warnings,
271 })
272}
273
274#[derive(Debug, Clone, PartialEq, Serialize)]
276pub struct DeployPlanResult {
277 pub env: String,
279 pub gateway: String,
281 pub mode: String,
283 pub node: String,
285 pub transport: String,
287 pub render_sha256: String,
289 pub plan_sha256: String,
291 pub summary: PlanSummary,
293 pub entries: Vec<PlanEntry>,
295 pub delete_set: DeleteSet,
298 pub base: Option<BaseSummary>,
300 pub warnings: Vec<String>,
302 pub excluded_tag_providers: Vec<String>,
306 pub unreadable: Vec<UnreadableUnit>,
311}
312
313#[derive(Debug, Clone, Default)]
315pub struct PlanOptions {
316 pub base: Option<PathBuf>,
319 pub diff: bool,
321}
322
323#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
325pub struct BaseSummary {
326 pub dir: String,
328 pub render_sha256: String,
330 pub files: usize,
332}
333
334pub async fn deploy_plan(
338 declared: &Declared,
339 reader: &dyn TreeReader,
340 opts: &PlanOptions,
341) -> Result<DeployPlanResult, CoreError> {
342 let node = &declared.target.nodes.master;
343 let transport = match (&node.ssh, &node.local) {
344 (Some(_), _) => "ssh",
345 (None, Some(_)) => "local",
346 (None, None) => "none",
347 };
348 let base = match &opts.base {
351 Some(dir) => {
352 let dir = dir.clone();
353 let (gateway, env, mode) = (
354 declared.target.gateway.clone(),
355 declared.env.clone(),
356 declared.target.mode.clone(),
357 );
358 let read =
359 tokio::task::spawn_blocking(move || BaseTree::read(&dir, &gateway, &env, &mode))
360 .await
361 .map_err(|e| CoreError::Internal(format!("--base read task failed: {e}")))?;
362 Some(read?)
363 }
364 None => None,
365 };
366 if let Some(journal) = journal_on_box(
371 reader,
372 node,
373 &declared.target.gateway,
374 &declared.target.mode,
375 )
376 .await?
377 {
378 return Err(journal_present(
379 &journal,
380 &declared.env,
381 node,
382 &declared.target.gateway,
383 &declared.target.mode,
384 ));
385 }
386 let excluded = declared.excluded_tag_providers();
387 let scope = Scope::from_manifest(&declared.manifest, &declared.target.mode, &excluded);
388 let observed = reader.read(&scope).await?;
389 let gateway = &declared.target.gateway;
390 let mode = &declared.target.mode;
391 let ledger = match reader.read_ledger(gateway, mode).await? {
394 Some(bytes) => Some(Ledger::parse(
395 &ledger_location(LedgerRoot::for_node(node), gateway, mode),
396 &bytes,
397 gateway,
398 mode,
399 )?),
400 None => None,
401 };
402 let mut planned = plan(
403 &declared.outcome.tree,
404 &observed,
405 ledger.as_ref(),
406 &scope,
407 &excluded,
408 );
409 if let Some(base) = &base {
410 apply_base(&mut planned, base);
411 }
412 if opts.diff {
413 render_diffs(&mut planned, &declared.outcome.tree, &observed);
414 }
415
416 let mut warnings = declared.outcome.warnings.clone();
417 warnings.extend(planned.warnings);
418 Ok(DeployPlanResult {
419 env: declared.env.clone(),
420 gateway: declared.target.gateway.clone(),
421 mode: declared.target.mode.clone(),
422 node: "master".to_string(),
423 transport: transport.to_string(),
424 render_sha256: declared.outcome.tree.render_sha256(),
425 plan_sha256: planned.plan_sha256,
426 summary: planned.summary,
427 entries: planned.entries,
428 delete_set: planned.delete_set,
429 base: base.map(|b| BaseSummary {
430 dir: b.dir,
431 render_sha256: b.render_sha256,
432 files: b.files.len(),
433 }),
434 warnings,
435 excluded_tag_providers: excluded,
436 unreadable: planned.unreadable,
437 })
438}
439
440#[derive(Debug, Clone, PartialEq, Serialize)]
442pub struct DeployPullResult {
443 pub env: String,
445 pub gateway: String,
447 pub mode: String,
449 pub node: String,
451 pub transport: String,
453 pub out: String,
455 pub files_written: usize,
458 pub manifest: String,
460 pub render_sha256: String,
463 pub skipped: Vec<Skipped>,
466 pub warnings: Vec<String>,
468 pub excluded_tag_providers: Vec<String>,
472}
473
474struct Existing {
478 files: Vec<String>,
479 dirs: BTreeSet<String>,
481}
482
483fn repo_entry_exists(repo: &Path, rel: &str) -> Result<bool, CoreError> {
485 match std::fs::symlink_metadata(repo.join(rel)) {
486 Ok(_) => Ok(true),
487 Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(false),
488 Err(e) => Err(CoreError::deploy_tree_invalid(
489 rel,
490 None,
491 None,
492 format!("cannot inspect ({})", e.kind()),
493 )),
494 }
495}
496
497fn scan_existing(repo: &Path, gateway: &str, env: &str) -> Result<Option<Existing>, CoreError> {
501 let gateway_rel = format!("gateways/{gateway}");
502 for rel in ["gateways", gateway_rel.as_str()] {
503 if !repo_entry_exists(repo, rel)? {
504 return Ok(None);
505 }
506 require_repo_dir(repo, rel)?;
507 }
508 let mut existing = Existing {
509 files: Vec::new(),
510 dirs: BTreeSet::new(),
511 };
512 for sub in [
513 "resources".to_string(),
514 "tags".to_string(),
515 format!("modes/{env}"),
516 ] {
517 if sub.starts_with("modes/") {
518 let modes = format!("{gateway_rel}/modes");
519 if !repo_entry_exists(repo, &modes)? {
520 continue;
521 }
522 require_repo_dir(repo, &modes)?;
523 }
524 let rel = format!("{gateway_rel}/{sub}");
525 if !repo_entry_exists(repo, &rel)? {
526 continue;
527 }
528 require_repo_dir(repo, &rel)?;
529 let mut files: BTreeMap<String, PathBuf> = BTreeMap::new();
530 let mut dirs: BTreeSet<String> = BTreeSet::new();
531 walk_capped(
532 &repo.join(&rel),
533 &sub,
534 &mut files,
535 &mut dirs,
536 READ_MAX_ENTRIES,
537 )
538 .map_err(|why| CoreError::deploy_tree_invalid(&rel, None, None, why))?;
539 existing.dirs.insert(sub.clone());
540 existing.dirs.extend(dirs);
541 existing.files.extend(files.into_keys());
542 }
543 Ok(Some(existing))
544}
545
546fn pull_prose(gateway: &str, env: &str, existing: &Existing) -> String {
548 let shown: Vec<&str> = existing.files.iter().take(5).map(String::as_str).collect();
549 let more = if existing.files.len() > shown.len() {
550 ", ..."
551 } else {
552 ""
553 };
554 format!(
555 "deploy pull would replace gateways/{gateway}/resources, tags and modes/{env} with the live tree of nodes.master ({} existing files, e.g. {}{more})",
556 existing.files.len(),
557 shown.join(", ")
558 )
559}
560
561pub fn deploy_pull_preview(repo: &Path, env: &str) -> Result<Option<String>, CoreError> {
566 deploy_pull_preview_in(repo, repo, env)
567}
568
569pub fn deploy_pull_preview_in(
572 repo: &Path,
573 targets: &Path,
574 env: &str,
575) -> Result<Option<String>, CoreError> {
576 let (_, target) = load_target(targets, env)?;
577 Ok(scan_existing(repo, &target.gateway, env)?
578 .filter(|existing| !existing.files.is_empty())
579 .map(|existing| pull_prose(&target.gateway, env, &existing)))
580}
581
582fn transport_label(node: &NodeTarget) -> &'static str {
583 match (&node.ssh, &node.local) {
584 (Some(_), _) => "ssh",
585 (None, Some(_)) => "local",
586 (None, None) => "none",
587 }
588}
589
590pub async fn deploy_pull(
596 repo: &Path,
597 env: &str,
598 reader: &dyn TreeReader,
599 yes: bool,
600) -> Result<DeployPullResult, CoreError> {
601 deploy_pull_in(repo, repo, env, reader, yes).await
602}
603
604pub async fn deploy_pull_in(
607 repo: &Path,
608 targets: &Path,
609 env: &str,
610 reader: &dyn TreeReader,
611 yes: bool,
612) -> Result<DeployPullResult, CoreError> {
613 let (_, target) = load_target(targets, env)?;
614 let gateway = target.gateway.clone();
615 let existing = scan_existing(repo, &gateway, env)?;
616 if let Some(existing) = existing.as_ref().filter(|e| !e.files.is_empty())
617 && !yes
618 {
619 return Err(CoreError::ConfirmationRequired {
620 operation: pull_prose(&gateway, env, existing),
621 });
622 }
623
624 let manifest_rel = format!("gateways/{gateway}/manifest.yaml");
627 let mut left_out: Vec<Skipped> = Vec::new();
628 let (manifest, generated, excluded) = if repo_entry_exists(repo, &manifest_rel)? {
629 let bytes = read_repo_file(repo, &manifest_rel, None)?;
630 let manifest = Manifest::parse(&manifest_rel, &bytes)?;
631 let excluded = exclude::effective(&target, Some(&manifest));
634 let contradictions = exclude::contradictions(&excluded, &manifest, &manifest_rel);
635 if !contradictions.is_empty() {
636 return Err(CoreError::deploy_findings(contradictions));
637 }
638 (manifest, None, excluded)
639 } else {
640 let excluded = exclude::effective(&target, None);
641 let mut live_types = reader.list_types(CORE_COLLECTION).await?;
642 live_types.extend(reader.list_types(&target.mode).await?);
643 let mut names = reader.list_tag_providers(CORE_COLLECTION).await?;
647 names.extend(reader.list_tag_providers(&target.mode).await?);
648 let providers: Vec<String> = names
649 .into_iter()
650 .filter(|p| !exclude::is_excluded(p, &excluded))
651 .collect::<BTreeSet<String>>()
652 .into_iter()
653 .collect();
654 let (yaml, skipped) = generate_manifest(&live_types, &providers);
655 left_out = skipped;
656 let manifest = Manifest::parse(&manifest_rel, yaml.as_bytes())?;
657 (manifest, Some(yaml), excluded)
658 };
659
660 let scope = Scope::from_manifest(&manifest, &target.mode, &excluded);
661 let observed = reader.read(&scope).await?;
662 let outcome = pull(&PullInput {
663 gateway: &gateway,
664 env,
665 target: &target,
666 manifest: &manifest,
667 observed: &observed,
668 })?;
669
670 write_pulled(
671 repo,
672 &gateway,
673 existing.as_ref(),
674 &outcome.source,
675 generated.as_deref(),
676 )?;
677
678 let removed = removed_unreadable_warnings(&gateway, env, existing.as_ref(), &outcome.skipped);
679 left_out.extend(outcome.skipped);
680 left_out.sort_by(|a, b| a.path.as_bytes().cmp(b.path.as_bytes()));
681 let mut warnings = outcome.rendered.warnings.clone();
682 warnings.extend(removed);
683 warnings.extend(observed.local_units.iter().map(|unit| {
684 format!(
685 "{unit} also exists in the `local` collection, which the gateway loads over the mode collection; pull did not read the `local` copy"
686 )
687 }));
688 Ok(DeployPullResult {
689 env: env.to_string(),
690 gateway: gateway.clone(),
691 mode: target.mode.clone(),
692 node: "master".to_string(),
693 transport: transport_label(&target.nodes.master).to_string(),
694 out: format!("gateways/{gateway}"),
695 files_written: outcome.source.files.len(),
696 manifest: if generated.is_some() {
697 "generated"
698 } else {
699 "kept"
700 }
701 .to_string(),
702 render_sha256: outcome.rendered.tree.render_sha256(),
703 skipped: left_out,
704 warnings,
705 excluded_tag_providers: excluded,
706 })
707}
708
709const REMOVED_FILES_SHOWN: usize = 10;
712
713fn removed_unreadable_warnings(
723 gateway: &str,
724 env: &str,
725 existing: Option<&Existing>,
726 skipped: &[Skipped],
727) -> Vec<String> {
728 let Some(existing) = existing else {
729 return Vec::new();
730 };
731 let mode_prefix = format!("modes/{env}/");
732 let mut out = Vec::new();
733 for unit in skipped.iter().filter(|s| s.reason == REASON_PATH_TOO_LONG) {
734 let under = format!("{}/", unit.path);
735 let removed: Vec<String> = existing
736 .files
737 .iter()
738 .filter(|file| {
739 let base = file.strip_prefix(&mode_prefix).unwrap_or(file);
740 RelPath::new(base)
741 .ok()
742 .and_then(|p| collection_path(&p).ok())
743 .is_some_and(|c| c.as_str() == unit.path || c.as_str().starts_with(&under))
744 })
745 .map(|file| format!("gateways/{gateway}/{file}"))
746 .collect();
747 if removed.is_empty() {
748 continue;
749 }
750 let take = removed.len().min(REMOVED_FILES_SHOWN);
751 let mut named = removed[..take].join(", ");
752 if removed.len() > take {
753 named.push_str(&format!(" and {} more", removed.len() - take));
754 }
755 out.push(format!(
756 "{} has a live path the host's sftp server cannot read ({REASON_PATH_TOO_LONG}), so pull did not write it and removed its {} repo file(s): {named}; the live unit is unchanged and git still holds the removed files",
757 unit.path,
758 removed.len()
759 ));
760 }
761 out
762}
763
764fn safe_component(name: &str) -> bool {
767 !name.is_empty()
768 && name != "."
769 && name != ".."
770 && !name.contains(['/', '\\', '\0'])
771 && !name.chars().any(char::is_control)
772}
773
774fn validate_pulled(gateway: &str, source: &SourceTree) -> Result<(), CoreError> {
780 let refuse = |path: &str, why: String| {
781 CoreError::deploy_tree_invalid(format!("gateways/{gateway}/{path}"), None, None, why)
782 };
783 let mut folded: BTreeMap<String, &str> = BTreeMap::new();
784 let mut dir_folds: BTreeSet<String> = BTreeSet::new();
785 for path in source.files.keys() {
786 let parts: Vec<&str> = path.components().collect();
787 if parts.iter().any(|c| !safe_component(c)) {
788 return Err(refuse(
789 path.as_str(),
790 "has an unsafe path component; refused before anything was changed".to_string(),
791 ));
792 }
793 let fold = path.as_str().to_lowercase();
794 if let Some(other) = folded.insert(fold.clone(), path.as_str()) {
795 return Err(refuse(
796 path.as_str(),
797 format!(
798 "collides with {other} when case is folded; a case-insensitive filesystem keeps only one. Refused before anything was changed"
799 ),
800 ));
801 }
802 for end in fold.match_indices('/').map(|(i, _)| i) {
803 dir_folds.insert(fold[..end].to_string());
804 }
805 }
806 for (fold, path) in &folded {
807 if dir_folds.contains(fold) {
808 return Err(refuse(
809 path,
810 "is both a file and a directory in the pulled tree; refused before anything was changed"
811 .to_string(),
812 ));
813 }
814 }
815 Ok(())
816}
817
818fn write_pulled(
822 repo: &Path,
823 gateway: &str,
824 existing: Option<&Existing>,
825 source: &SourceTree,
826 manifest: Option<&str>,
827) -> Result<(), CoreError> {
828 validate_pulled(gateway, source)?;
829 write_pulled_fs(repo, gateway, existing, source, manifest)
830}
831
832#[cfg(unix)]
836fn write_pulled_fs(
837 repo: &Path,
838 gateway: &str,
839 existing: Option<&Existing>,
840 source: &SourceTree,
841 manifest: Option<&str>,
842) -> Result<(), CoreError> {
843 use crate::deploy::tree::dirfd::{self, FdError, Kind};
844 use std::io::Write;
845
846 let fail = |what: &str, e: FdError| match e {
847 FdError::Changed => CoreError::deploy_tree_invalid(
848 what,
849 None,
850 None,
851 "changed while it was written (a symlink or a non-directory appeared); refused, nothing outside the gateway directory was touched",
852 ),
853 FdError::Io(io) => CoreError::Internal(format!("cannot write {what}: {io}")),
854 other => CoreError::Internal(format!("cannot write {what}: {other:?}")),
855 };
856
857 let repo_real = std::fs::canonicalize(repo).map_err(|e| {
861 CoreError::Internal(format!("cannot resolve the repo {}: {e}", repo.display()))
862 })?;
863 let root = dirfd::open_root(&repo_real).map_err(|e| fail("the repo", e))?;
864 let gateways = dirfd::ensure_dir(&root, "gateways").map_err(|e| fail("gateways", e))?;
865 let gw_label = format!("gateways/{gateway}");
866 let gw = dirfd::ensure_dir(&gateways, gateway).map_err(|e| fail(&gw_label, e))?;
867
868 if let Some(existing) = existing {
869 for file in &existing.files {
870 let parts: Vec<&str> = file.split('/').collect();
871 let (name, parents) = parts.split_last().ok_or_else(|| {
872 CoreError::Internal("an existing file has an empty path".to_string())
873 })?;
874 let label = format!("{gw_label}/{file}");
875 let dir = dirfd::walk(&gw, parents).map_err(|e| fail(&label, e))?;
876 match dirfd::kind_at(&dir, name).map_err(|e| fail(&label, e))? {
877 Kind::File => dirfd::unlink_at(&dir, name).map_err(|e| fail(&label, e))?,
878 Kind::Missing => {}
879 _ => return Err(fail(&label, FdError::Changed)),
880 }
881 }
882 for dir_path in existing.dirs.iter().rev() {
884 let parts: Vec<&str> = dir_path.split('/').collect();
885 let Some((name, parents)) = parts.split_last() else {
886 continue;
887 };
888 let label = format!("{gw_label}/{dir_path}");
889 let parent = dirfd::walk(&gw, parents).map_err(|e| fail(&label, e))?;
890 match dirfd::rmdir_at(&parent, name) {
891 Ok(()) | Err(FdError::Missing) => {}
892 Err(FdError::NotEmpty) => {
893 return Err(CoreError::deploy_tree_invalid(
894 label,
895 None,
896 None,
897 "is not empty after the pull removed its files (something wrote into it); nothing more was changed",
898 ));
899 }
900 Err(e) => return Err(fail(&label, e)),
901 }
902 }
903 }
904
905 if let Some(text) = manifest {
906 let label = format!("{gw_label}/manifest.yaml");
907 let mut file = dirfd::create_new(&gw, "manifest.yaml").map_err(|e| fail(&label, e))?;
908 file.write_all(text.as_bytes())
909 .map_err(|e| CoreError::Internal(format!("cannot write {label}: {e}")))?;
910 }
911 for (path, bytes) in &source.files {
912 let label = format!("{gw_label}/{path}");
913 let parts: Vec<&str> = path.components().collect();
914 if parts.iter().any(|c| !safe_component(c)) {
915 return Err(CoreError::deploy_tree_invalid(
916 label,
917 None,
918 None,
919 "has an unsafe path component; refused",
920 ));
921 }
922 let (name, parents) = parts
923 .split_last()
924 .ok_or_else(|| CoreError::Internal("a pulled file has an empty path".to_string()))?;
925 let dir = dirfd::ensure_walk(&gw, parents).map_err(|e| fail(&label, e))?;
926 let mut file = dirfd::create_new(&dir, name).map_err(|e| fail(&label, e))?;
927 file.write_all(bytes)
928 .map_err(|e| CoreError::Internal(format!("cannot write {label}: {e}")))?;
929 }
930 Ok(())
931}
932
933#[cfg(not(unix))]
937fn write_pulled_fs(
938 repo: &Path,
939 gateway: &str,
940 existing: Option<&Existing>,
941 source: &SourceTree,
942 manifest: Option<&str>,
943) -> Result<(), CoreError> {
944 use std::io::Write;
945
946 let io = |what: &Path, e: std::io::Error| {
947 CoreError::Internal(format!("cannot write {}: {e}", what.display()))
948 };
949 let gw = repo.join("gateways").join(gateway);
950 std::fs::create_dir_all(&gw).map_err(|e| io(&gw, e))?;
951 if let Some(existing) = existing {
952 for file in &existing.files {
953 let path = file.split('/').fold(gw.clone(), |acc, c| acc.join(c));
954 std::fs::remove_file(&path).map_err(|e| io(&path, e))?;
955 }
956 for dir_path in existing.dirs.iter().rev() {
957 let path = dir_path.split('/').fold(gw.clone(), |acc, c| acc.join(c));
958 std::fs::remove_dir(&path).map_err(|e| io(&path, e))?;
959 }
960 }
961 let create = |path: &Path, bytes: &[u8]| -> Result<(), CoreError> {
962 if let Some(parent) = path.parent() {
963 std::fs::create_dir_all(parent).map_err(|e| io(parent, e))?;
964 }
965 let mut file = std::fs::OpenOptions::new()
966 .write(true)
967 .create_new(true)
968 .open(path)
969 .map_err(|e| io(path, e))?;
970 file.write_all(bytes).map_err(|e| io(path, e))
971 };
972 if let Some(text) = manifest {
973 create(&gw.join("manifest.yaml"), text.as_bytes())?;
974 }
975 for (path, bytes) in &source.files {
976 let parts: Vec<&str> = path.components().collect();
977 if parts.iter().any(|c| !safe_component(c)) {
978 return Err(CoreError::deploy_tree_invalid(
979 format!("gateways/{gateway}/{path}"),
980 None,
981 None,
982 "has an unsafe path component; refused",
983 ));
984 }
985 let dest = parts.iter().fold(gw.clone(), |acc, c| acc.join(c));
986 create(&dest, bytes)?;
987 }
988 Ok(())
989}
990
991#[derive(Debug, Clone)]
993pub struct VerifyOptions {
994 pub probe_deadline: Duration,
996 pub probe_interval: Duration,
998 pub log_settle: Duration,
1001}
1002
1003impl Default for VerifyOptions {
1004 fn default() -> Self {
1005 Self {
1006 probe_deadline: verify::PROBE_DEADLINE,
1007 probe_interval: verify::PROBE_INTERVAL,
1008 log_settle: Duration::from_millis(verify::LOG_SETTLE_MS.unsigned_abs()),
1009 }
1010 }
1011}
1012
1013#[derive(Debug, Clone, PartialEq, Serialize)]
1015pub struct DeployVerifyResult {
1016 pub env: String,
1018 pub gateway: String,
1020 pub mode: String,
1022 pub node: String,
1025 pub project: String,
1027 pub ok: bool,
1029 pub slug: Option<&'static str>,
1031 pub checks: Vec<verify::Check>,
1033 pub resources_checked: usize,
1035 pub warnings: Vec<String>,
1037 pub excluded_tag_providers: Vec<String>,
1040 #[serde(skip_serializing_if = "Option::is_none")]
1043 pub backup: Option<BackupReport>,
1044 #[serde(skip_serializing_if = "Option::is_none")]
1047 pub secrets: Option<SecretNamesReport>,
1048}
1049
1050impl DeployVerifyResult {
1051 pub fn failure_exit_code(&self) -> Option<u8> {
1056 (!self.ok).then_some(6)
1057 }
1058}
1059
1060const LOG_PAGES_MAX: usize = 5;
1062
1063pub async fn deploy_verify(
1078 declared: &Declared,
1079 project: &str,
1080 deploy: &dyn DeployApi,
1081 gateway: &dyn GatewayApi,
1082 reader: &dyn TreeReader,
1083 opts: &VerifyOptions,
1084) -> Result<DeployVerifyResult, CoreError> {
1085 let expected =
1086 verify::Expected::from_render(&declared.target.mode, &declared.outcome, &declared.manifest);
1087
1088 let scan = deploy.scan_status(ScanKind::Config).await?;
1089 let scan_read_at = std::time::Instant::now();
1090 let mode = normalize_deployment_mode(deploy.deployment_mode().await?);
1091 let read_back = gather_read_back(deploy, &expected).await?;
1092
1093 let excluded = declared.excluded_tag_providers();
1096 let scope = Scope::from_manifest(&declared.manifest, &declared.target.mode, &excluded);
1097 let live = reader.read(&scope).await?;
1098 let planned = plan(&declared.outcome.tree, &live, None, &scope, &excluded);
1099 let files = planned
1100 .entries
1101 .iter()
1102 .filter(|entry| expected.tag_units.contains(&entry.key))
1103 .map(|entry| (entry.key.clone(), entry.status))
1104 .collect();
1105
1106 let logs = if scan.scan_active {
1107 Ok(Vec::new())
1109 } else {
1110 settle_logs(gateway, &scan, scan_read_at, opts.log_settle).await?;
1111 read_logs(gateway, verify::log_window_start(&scan)).await?
1112 };
1113 let probes = if expected.probes.is_empty() {
1114 None
1115 } else {
1116 Some(poll_probes(gateway, project, &expected.probes, opts).await?)
1117 };
1118
1119 let observed = verify::ObservedState {
1120 scan,
1121 mode,
1122 read_back,
1123 files,
1124 logs,
1125 probes,
1126 unreadable: planned
1127 .unreadable
1128 .iter()
1129 .map(|unit| (unit.key.clone(), unit.declared))
1130 .collect(),
1131 };
1132 let report = verify::evaluate(&expected, &observed);
1133 Ok(DeployVerifyResult {
1134 env: declared.env.clone(),
1135 gateway: declared.target.gateway.clone(),
1136 mode: declared.target.mode.clone(),
1137 node: "master".to_string(),
1138 project: project.to_string(),
1139 ok: report.ok,
1140 slug: report.slug,
1141 checks: report.checks,
1142 resources_checked: report.resources_checked,
1143 warnings: report.warnings,
1144 excluded_tag_providers: excluded,
1145 backup: None,
1146 secrets: None,
1147 })
1148}
1149
1150#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1152pub enum VerifyNode {
1153 Master,
1156 Backup,
1158 Both,
1160}
1161
1162impl VerifyNode {
1163 pub fn as_str(self) -> &'static str {
1165 match self {
1166 VerifyNode::Master => "master",
1167 VerifyNode::Backup => "backup",
1168 VerifyNode::Both => "both",
1169 }
1170 }
1171}
1172
1173pub type VerifyMasterSurfaces<'a> = (&'a dyn DeployApi, &'a dyn GatewayApi);
1175
1176pub async fn deploy_verify_node(
1190 declared: &Declared,
1191 project: &str,
1192 node: VerifyNode,
1193 master: Option<VerifyMasterSurfaces<'_>>,
1194 master_reader: &dyn TreeReader,
1195 backup: Option<&BackupNode<'_>>,
1196 opts: &VerifyOptions,
1197) -> Result<DeployVerifyResult, CoreError> {
1198 let missing = |what: &str| CoreError::InvalidInput {
1199 reason: format!("deploy verify --node {} needs the {what}", node.as_str()),
1200 };
1201 let master_result = if node == VerifyNode::Backup {
1202 None
1203 } else {
1204 let (deploy, gateway) = master.ok_or_else(|| missing("master surfaces"))?;
1205 Some(deploy_verify(declared, project, deploy, gateway, master_reader, opts).await?)
1206 };
1207 if node == VerifyNode::Master {
1208 return master_result.ok_or_else(|| missing("master surfaces"));
1209 }
1210 let backup = backup.ok_or_else(|| missing("backup node"))?;
1211 let expected =
1212 verify::Expected::from_render(&declared.target.mode, &declared.outcome, &declared.manifest);
1213 let mut backup_warnings = Vec::new();
1214 let (tick, compared) = verify_backup_tick(
1215 declared,
1216 &expected,
1217 master_reader,
1218 backup,
1219 &mut backup_warnings,
1220 )
1221 .await;
1222 let report = verify_report(&tick);
1223
1224 let mut result = match master_result {
1225 Some(mut result) => {
1229 result.warnings.extend(backup_warnings);
1230 result
1231 }
1232 None => DeployVerifyResult {
1233 env: declared.env.clone(),
1234 gateway: declared.target.gateway.clone(),
1235 mode: declared.target.mode.clone(),
1236 node: String::new(),
1237 project: project.to_string(),
1238 ok: true,
1239 slug: None,
1240 checks: Vec::new(),
1241 resources_checked: compared,
1242 warnings: backup_warnings,
1243 excluded_tag_providers: declared.excluded_tag_providers(),
1244 backup: None,
1245 secrets: None,
1246 },
1247 };
1248 let secrets = verify_secret_names(
1251 declared,
1252 master
1253 .map(|(deploy, _)| deploy)
1254 .filter(|_| node != VerifyNode::Backup),
1255 Some(backup),
1256 )
1257 .await?;
1258 let secrets_ok = secrets.nodes.iter().all(|n| n.status != CheckStatus::Fail);
1259 result.node = node.as_str().to_string();
1260 result.ok = result.ok && report.converged && secrets_ok;
1261 result.slug = (!result.ok).then_some(verify::VERIFY_FAILED_SLUG);
1262 result.backup = Some(report);
1263 result.secrets = Some(secrets);
1264 Ok(result)
1265}
1266
1267async fn verify_secret_names(
1274 declared: &Declared,
1275 master: Option<&dyn DeployApi>,
1276 backup: Option<&BackupNode<'_>>,
1277) -> Result<SecretNamesReport, CoreError> {
1278 let referenced = collect_referenced_secrets(&declared.outcome.tree);
1279 let declared_providers = declared_secret_providers(&declared.outcome.tree);
1280 let providers: BTreeSet<&str> = referenced.iter().map(|(p, _)| p.as_str()).collect();
1281 let mut nodes = Vec::with_capacity(2);
1282 if let Some(master) = master {
1283 let mut listings = BTreeMap::new();
1284 for provider in &providers {
1285 listings.insert(
1286 (*provider).to_string(),
1287 master.secret_names(provider).await?,
1288 );
1289 }
1290 nodes.push(check_secret_names(
1291 "master",
1292 &referenced,
1293 &listings,
1294 &declared_providers,
1295 SecretCheckMode::Verify,
1296 ));
1297 }
1298 if let Some(backup) = backup {
1299 let mut listings = BTreeMap::new();
1300 for provider in &providers {
1301 listings.insert(
1302 (*provider).to_string(),
1303 backup.secret_names(provider).await?,
1304 );
1305 }
1306 nodes.push(check_secret_names(
1307 "backup",
1308 &referenced,
1309 &listings,
1310 &declared_providers,
1311 SecretCheckMode::Verify,
1312 ));
1313 }
1314 Ok(SecretNamesReport::ordered(nodes))
1315}
1316
1317async fn verify_backup_tick(
1324 declared: &Declared,
1325 expected: &verify::Expected,
1326 master_reader: &dyn TreeReader,
1327 backup: &BackupNode<'_>,
1328 warnings: &mut Vec<String>,
1329) -> (TickOutcome, usize) {
1330 let signals = match backup.status().await {
1331 Ok(status) => Ok(BackupSignals::from_status(None, &status)),
1332 Err(error) => Err(error.code().to_string()),
1333 };
1334 let mode = match backup.deployment_mode().await {
1335 Ok(mode) => mode_backup_check(expected, mode.as_deref()),
1336 Err(error) => unreachable_check(BackupCheckId::Mode, error.code()),
1337 };
1338 let mut compared = 0;
1339 let read_back = match backup.read_back(expected).await {
1340 Ok(read_back) => {
1341 let (check, count) = read_back_backup_check(expected, &read_back, warnings);
1342 compared = count;
1343 check
1344 }
1345 Err(error) => unreachable_check(BackupCheckId::ReadBack, error.code()),
1346 };
1347 let (files, files_checked) =
1348 verify_backup_files(declared, master_reader, backup, warnings).await;
1349 (
1350 TickOutcome {
1351 signals,
1352 exhausted: false,
1353 content: Some(vec![mode, read_back, files]),
1354 files_checked,
1355 stable_ticks: 0,
1356 },
1357 compared,
1358 )
1359}
1360
1361async fn verify_backup_files(
1366 declared: &Declared,
1367 master_reader: &dyn TreeReader,
1368 backup: &BackupNode<'_>,
1369 warnings: &mut Vec<String>,
1370) -> (BackupCheck, usize) {
1371 let scope = Scope::from_manifest(
1372 &declared.manifest,
1373 &declared.target.mode,
1374 &declared.excluded_tag_providers(),
1375 );
1376 let backup_files = match backup.read_files(&scope).await {
1377 None => return (files_skipped_check(), 0),
1379 Some(Ok(observed)) => observed,
1380 Some(Err(error)) => {
1381 return (
1382 unreachable_check(BackupCheckId::Files, &format!("backup {}", error.code())),
1383 0,
1384 );
1385 }
1386 };
1387 let backup_unreadable = unreadable_units(&backup_files);
1388 let backup_files = effective_files(backup_files);
1389 let master_files = match master_reader.read(&scope).await {
1390 Ok(observed) => observed,
1391 Err(error) => {
1392 return (
1393 unreachable_check(BackupCheckId::Files, &format!("master {}", error.code())),
1394 0,
1395 );
1396 }
1397 };
1398 let master_unreadable = unreadable_units(&master_files);
1399 let master_files = effective_files(master_files);
1400 let present: BTreeSet<_> = declared.outcome.tree.files.keys().cloned().collect();
1401 let absent = BTreeSet::new();
1402 let mut comparison = compare_files(&master_files, &backup_files, &present, &absent);
1403 exempt_unreadable(
1404 &mut comparison,
1405 &present,
1406 &absent,
1407 &[
1408 ("master", &master_unreadable),
1409 ("backup", &backup_unreadable),
1410 ],
1411 );
1412 warnings.extend(unreadable_warnings(&comparison));
1413 (files_check(&comparison), comparison.checked)
1414}
1415
1416pub(crate) async fn gather_read_back(
1421 deploy: &dyn DeployApi,
1422 expected: &verify::Expected,
1423) -> Result<BTreeMap<String, verify::Remote>, CoreError> {
1424 let classify = |record: crate::client::config_resources::ResourceRecord| {
1425 if record.collection.as_deref() == Some(verify::SYSTEM_COLLECTION) {
1426 verify::Remote::System
1427 } else {
1428 verify::Remote::Config(record.config)
1429 }
1430 };
1431 let mut lists: BTreeMap<(String, String), Option<Vec<_>>> = BTreeMap::new();
1432 let mut out = BTreeMap::new();
1433 for res in &expected.resources {
1434 let remote = match &res.name {
1435 None => match deploy.resource_singleton(&res.module, &res.ty, None).await {
1436 Ok(record) => classify(record),
1437 Err(CoreError::NotFound { .. }) => verify::Remote::Missing,
1438 Err(e) => return Err(e),
1439 },
1440 Some(name) => {
1441 let type_key = (res.module.clone(), res.ty.clone());
1442 if !lists.contains_key(&type_key) {
1443 let items = match deploy
1444 .resource_list(&res.module, &res.ty, &ListQuery::default())
1445 .await
1446 {
1447 Ok(envelope) => Some(envelope.items),
1448 Err(CoreError::NotFound { .. }) => None,
1449 Err(e) => return Err(e),
1450 };
1451 lists.insert(type_key.clone(), items);
1452 }
1453 lists
1454 .get(&type_key)
1455 .and_then(Option::as_ref)
1456 .and_then(|items| items.iter().find(|r| r.name.as_deref() == Some(name)))
1457 .map_or(verify::Remote::Missing, |r| classify(r.clone()))
1458 }
1459 };
1460 out.insert(res.key.clone(), remote);
1461 }
1462 Ok(out)
1463}
1464
1465async fn settle_logs(
1469 gateway: &dyn GatewayApi,
1470 scan: &ScanStatus,
1471 scan_read_at: std::time::Instant,
1472 settle: Duration,
1473) -> Result<(), CoreError> {
1474 if settle.is_zero() {
1475 return Ok(());
1476 }
1477 let settle_ms = i64::try_from(settle.as_millis()).unwrap_or(i64::MAX);
1478 let probe = LogQuery {
1479 start_time: Some(scan.last_scan_timestamp.saturating_add(settle_ms)),
1480 limit: 1,
1481 ..LogQuery::default()
1482 };
1483 match gateway.logs(&probe).await {
1484 Ok(envelope) if !envelope.items.is_empty() => return Ok(()),
1485 Err(e @ CoreError::Network { .. }) => return Err(e),
1486 Ok(_) | Err(_) => {}
1488 }
1489 tokio::time::sleep(settle.saturating_sub(scan_read_at.elapsed())).await;
1490 Ok(())
1491}
1492
1493async fn read_logs(
1497 gateway: &dyn GatewayApi,
1498 start_time: i64,
1499) -> Result<Result<Vec<verify::LogLine>, String>, CoreError> {
1500 let mut lines = Vec::new();
1501 let mut offset = 0i64;
1502 for _ in 0..LOG_PAGES_MAX {
1503 let query = LogQuery {
1504 start_time: Some(start_time),
1505 min_level: Some("WARN".to_string()),
1506 limit: DEFAULT_LOG_LIMIT,
1507 offset,
1508 ..LogQuery::default()
1509 };
1510 match gateway.logs(&query).await {
1511 Ok(envelope) => {
1512 let count = i64::try_from(envelope.items.len()).unwrap_or(i64::MAX);
1513 lines.extend(envelope.items.into_iter().map(|e| verify::LogLine {
1514 timestamp: e.timestamp,
1515 logger: e.logger_name,
1516 message: e.message,
1517 }));
1518 if count < DEFAULT_LOG_LIMIT {
1519 return Ok(Ok(lines));
1520 }
1521 offset += count;
1522 }
1523 Err(e @ CoreError::Network { .. }) => return Err(e),
1524 Err(e) => return Ok(Err(e.code().to_string())),
1525 }
1526 }
1527 Ok(Err("logs_truncated".to_string()))
1528}
1529
1530async fn poll_probes(
1535 gateway: &dyn GatewayApi,
1536 project: &str,
1537 paths: &[String],
1538 opts: &VerifyOptions,
1539) -> Result<Result<Vec<verify::ProbeReading>, String>, CoreError> {
1540 let interval = opts.probe_interval.max(Duration::from_millis(1));
1541 let started = tokio::time::Instant::now();
1542 loop {
1543 let readings: Vec<verify::ProbeReading> =
1544 match crate::actions::tags::tags_read(gateway, project, paths).await {
1545 Ok(result) => result
1546 .results
1547 .into_iter()
1548 .enumerate()
1549 .map(|(i, row)| verify::ProbeReading {
1550 path: if row.path.is_empty() {
1552 paths.get(i).cloned().unwrap_or_default()
1553 } else {
1554 row.path
1555 },
1556 quality: row.quality,
1557 })
1558 .collect(),
1559 Err(e @ CoreError::Network { .. }) => return Err(e),
1560 Err(e) => return Ok(Err(e.code().to_string())),
1561 };
1562 let all_good = paths.iter().all(|path| {
1563 readings
1564 .iter()
1565 .any(|r| &r.path == path && r.quality.starts_with("Good"))
1566 });
1567 if all_good || started.elapsed() + interval > opts.probe_deadline {
1568 return Ok(Ok(readings));
1569 }
1570 tokio::time::sleep(interval).await;
1571 }
1572}
1573
1574#[cfg(test)]
1575mod tests {
1576 use super::*;
1577
1578 fn repo_with_existing() -> (tempfile::TempDir, Existing) {
1580 let repo = tempfile::TempDir::new().expect("repo");
1581 let dir = repo.path().join("gateways/g/resources/ignition/x");
1582 std::fs::create_dir_all(&dir).expect("dirs");
1583 std::fs::write(dir.join("keep.json"), b"{\"keep\": true}").expect("file");
1584 let existing = Existing {
1585 files: vec!["resources/ignition/x/keep.json".to_string()],
1586 dirs: ["resources", "resources/ignition", "resources/ignition/x"]
1587 .iter()
1588 .map(|d| d.to_string())
1589 .collect(),
1590 };
1591 (repo, existing)
1592 }
1593
1594 fn snapshot(root: &Path) -> BTreeMap<PathBuf, Vec<u8>> {
1595 fn walk(dir: &Path, out: &mut BTreeMap<PathBuf, Vec<u8>>) {
1596 for entry in std::fs::read_dir(dir).expect("read_dir") {
1597 let path = entry.expect("entry").path();
1598 if path.is_dir() {
1599 out.insert(path.clone(), Vec::new());
1600 walk(&path, out);
1601 } else {
1602 out.insert(path.clone(), std::fs::read(&path).expect("read"));
1603 }
1604 }
1605 }
1606 let mut out = BTreeMap::new();
1607 walk(root, &mut out);
1608 out
1609 }
1610
1611 fn assert_refused_untouched(files: &[(&str, &str)], needle: &str) {
1614 let (repo, existing) = repo_with_existing();
1615 let before = snapshot(repo.path());
1616 let source = SourceTree::from_files(files.iter().copied()).expect("source");
1617 let err =
1618 write_pulled(repo.path(), "g", Some(&existing), &source, None).expect_err("refused");
1619 assert_eq!(err.code(), "deploy_tree_invalid", "{err}");
1620 assert!(err.to_string().contains(needle), "{needle}: {err}");
1621 assert_eq!(
1622 snapshot(repo.path()),
1623 before,
1624 "a refused pull must not delete the existing tree"
1625 );
1626 }
1627
1628 #[test]
1629 fn unsafe_component_is_refused_before_any_delete() {
1630 assert_refused_untouched(
1631 &[
1632 ("resources/ignition/x/fine.json", "{}"),
1633 ("resources/ignition/x/bad\nname.json", "{}"),
1634 ],
1635 "unsafe path component",
1636 );
1637 }
1638
1639 #[test]
1640 fn case_fold_collision_is_refused_before_any_delete() {
1641 assert_refused_untouched(
1642 &[
1643 ("resources/ignition/x/Foo.json", "{}"),
1644 ("resources/ignition/x/foo.json", "{}"),
1645 ],
1646 "case is folded",
1647 );
1648 }
1649
1650 #[test]
1651 fn file_and_directory_clash_is_refused_before_any_delete() {
1652 assert_refused_untouched(
1653 &[
1654 ("resources/ignition/x", "{}"),
1655 ("resources/ignition/x/a.json", "{}"),
1656 ],
1657 "both a file and a directory",
1658 );
1659 }
1660
1661 #[test]
1662 fn a_valid_tree_replaces_the_existing_one() {
1663 let (repo, existing) = repo_with_existing();
1664 let source =
1665 SourceTree::from_files([("resources/ignition/x/new.json", "{}")]).expect("source");
1666 write_pulled(repo.path(), "g", Some(&existing), &source, None).expect("written");
1667 let dir = repo.path().join("gateways/g/resources/ignition/x");
1668 assert!(!dir.join("keep.json").exists());
1669 assert!(dir.join("new.json").is_file());
1670 }
1671}