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