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_with, unreachable_check,
103 unreadable_units, unreadable_warnings, verify_report,
104};
105use crate::deploy::plan::{
106 BaseTree, PlanEntry, PlanStatus, PlanSummary, UnreadableUnit, apply_base,
107 live_tag_folder_findings, pair_tag_keys, plan, 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 if declared.target.nodes.backup.is_some() {
427 warnings.extend(pair_tags_warning(
428 &planned.entries,
429 &planned.delete_set,
430 mode,
431 ));
432 }
433 Ok(DeployPlanResult {
434 env: declared.env.clone(),
435 gateway: declared.target.gateway.clone(),
436 mode: declared.target.mode.clone(),
437 node: "master".to_string(),
438 transport: transport.to_string(),
439 render_sha256: declared.outcome.tree.render_sha256(),
440 plan_sha256: planned.plan_sha256,
441 summary: planned.summary,
442 entries: planned.entries,
443 delete_set: planned.delete_set,
444 base: base.map(|b| BaseSummary {
445 dir: b.dir,
446 render_sha256: b.render_sha256,
447 files: b.files.len(),
448 }),
449 warnings,
450 excluded_tag_providers: excluded,
451 unreadable: planned.unreadable,
452 })
453}
454
455fn pair_tags_warning(entries: &[PlanEntry], delete: &DeleteSet, mode: &str) -> Option<String> {
461 let keys = pair_tag_keys(entries, delete);
462 if keys.is_empty() {
463 return None;
464 }
465 let shown = keys.len().min(crate::error::APPLY_LIST_SHOWN);
466 let mut named = keys[..shown].join(", ");
467 if keys.len() > shown {
468 named.push_str(&format!(" and {} more", keys.len() - shown));
469 }
470 Some(format!(
471 "redundant pair: {} tag unit(s) would be written to or kept in the {mode} collection, \
472 which Ignition's redundancy sync cannot carry; deploy apply refuses this plan \
473 (deploy_pair_tags_refused): {named}",
474 keys.len()
475 ))
476}
477
478#[derive(Debug, Clone, PartialEq, Serialize)]
480pub struct DeployPullResult {
481 pub env: String,
483 pub gateway: String,
485 pub mode: String,
487 pub node: String,
489 pub transport: String,
491 pub out: String,
493 pub files_written: usize,
496 pub manifest: String,
498 pub render_sha256: String,
501 pub skipped: Vec<Skipped>,
504 pub warnings: Vec<String>,
506 pub excluded_tag_providers: Vec<String>,
510}
511
512struct Existing {
516 files: Vec<String>,
517 dirs: BTreeSet<String>,
519}
520
521fn repo_entry_exists(repo: &Path, rel: &str) -> Result<bool, CoreError> {
523 match std::fs::symlink_metadata(repo.join(rel)) {
524 Ok(_) => Ok(true),
525 Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(false),
526 Err(e) => Err(CoreError::deploy_tree_invalid(
527 rel,
528 None,
529 None,
530 format!("cannot inspect ({})", e.kind()),
531 )),
532 }
533}
534
535fn scan_existing(repo: &Path, gateway: &str, env: &str) -> Result<Option<Existing>, CoreError> {
539 let gateway_rel = format!("gateways/{gateway}");
540 for rel in ["gateways", gateway_rel.as_str()] {
541 if !repo_entry_exists(repo, rel)? {
542 return Ok(None);
543 }
544 require_repo_dir(repo, rel)?;
545 }
546 let mut existing = Existing {
547 files: Vec::new(),
548 dirs: BTreeSet::new(),
549 };
550 for sub in [
551 "resources".to_string(),
552 "tags".to_string(),
553 format!("modes/{env}"),
554 ] {
555 if sub.starts_with("modes/") {
556 let modes = format!("{gateway_rel}/modes");
557 if !repo_entry_exists(repo, &modes)? {
558 continue;
559 }
560 require_repo_dir(repo, &modes)?;
561 }
562 let rel = format!("{gateway_rel}/{sub}");
563 if !repo_entry_exists(repo, &rel)? {
564 continue;
565 }
566 require_repo_dir(repo, &rel)?;
567 let mut files: BTreeMap<String, PathBuf> = BTreeMap::new();
568 let mut dirs: BTreeSet<String> = BTreeSet::new();
569 walk_capped(
570 &repo.join(&rel),
571 &sub,
572 &mut files,
573 &mut dirs,
574 READ_MAX_ENTRIES,
575 )
576 .map_err(|why| CoreError::deploy_tree_invalid(&rel, None, None, why))?;
577 existing.dirs.insert(sub.clone());
578 existing.dirs.extend(dirs);
579 existing.files.extend(files.into_keys());
580 }
581 Ok(Some(existing))
582}
583
584fn pull_prose(gateway: &str, env: &str, existing: &Existing) -> String {
586 let shown: Vec<&str> = existing.files.iter().take(5).map(String::as_str).collect();
587 let more = if existing.files.len() > shown.len() {
588 ", ..."
589 } else {
590 ""
591 };
592 format!(
593 "deploy pull would replace gateways/{gateway}/resources, tags and modes/{env} with the live tree of nodes.master ({} existing files, e.g. {}{more})",
594 existing.files.len(),
595 shown.join(", ")
596 )
597}
598
599pub fn deploy_pull_preview(repo: &Path, env: &str) -> Result<Option<String>, CoreError> {
604 deploy_pull_preview_in(repo, repo, env)
605}
606
607pub fn deploy_pull_preview_in(
610 repo: &Path,
611 targets: &Path,
612 env: &str,
613) -> Result<Option<String>, CoreError> {
614 let (_, target) = load_target(targets, env)?;
615 Ok(scan_existing(repo, &target.gateway, env)?
616 .filter(|existing| !existing.files.is_empty())
617 .map(|existing| pull_prose(&target.gateway, env, &existing)))
618}
619
620fn transport_label(node: &NodeTarget) -> &'static str {
621 match (&node.ssh, &node.local) {
622 (Some(_), _) => "ssh",
623 (None, Some(_)) => "local",
624 (None, None) => "none",
625 }
626}
627
628pub async fn deploy_pull(
634 repo: &Path,
635 env: &str,
636 reader: &dyn TreeReader,
637 yes: bool,
638) -> Result<DeployPullResult, CoreError> {
639 deploy_pull_in(repo, repo, env, reader, yes).await
640}
641
642pub async fn deploy_pull_in(
645 repo: &Path,
646 targets: &Path,
647 env: &str,
648 reader: &dyn TreeReader,
649 yes: bool,
650) -> Result<DeployPullResult, CoreError> {
651 let (_, target) = load_target(targets, env)?;
652 let gateway = target.gateway.clone();
653 let existing = scan_existing(repo, &gateway, env)?;
654 if let Some(existing) = existing.as_ref().filter(|e| !e.files.is_empty())
655 && !yes
656 {
657 return Err(CoreError::ConfirmationRequired {
658 operation: pull_prose(&gateway, env, existing),
659 });
660 }
661
662 let manifest_rel = format!("gateways/{gateway}/manifest.yaml");
665 let mut left_out: Vec<Skipped> = Vec::new();
666 let (manifest, generated, excluded) = if repo_entry_exists(repo, &manifest_rel)? {
667 let bytes = read_repo_file(repo, &manifest_rel, None)?;
668 let manifest = Manifest::parse(&manifest_rel, &bytes)?;
669 let excluded = exclude::effective(&target, Some(&manifest));
672 let contradictions = exclude::contradictions(&excluded, &manifest, &manifest_rel);
673 if !contradictions.is_empty() {
674 return Err(CoreError::deploy_findings(contradictions));
675 }
676 (manifest, None, excluded)
677 } else {
678 let excluded = exclude::effective(&target, None);
679 let mut live_types = reader.list_types(CORE_COLLECTION).await?;
680 live_types.extend(reader.list_types(&target.mode).await?);
681 let mut names = reader.list_tag_providers(CORE_COLLECTION).await?;
685 names.extend(reader.list_tag_providers(&target.mode).await?);
686 let providers: Vec<String> = names
687 .into_iter()
688 .filter(|p| !exclude::is_excluded(p, &excluded))
689 .collect::<BTreeSet<String>>()
690 .into_iter()
691 .collect();
692 let (yaml, skipped) = generate_manifest(&live_types, &providers);
693 left_out = skipped;
694 let manifest = Manifest::parse(&manifest_rel, yaml.as_bytes())?;
695 (manifest, Some(yaml), excluded)
696 };
697
698 let scope = Scope::from_manifest(&manifest, &target.mode, &excluded);
699 let observed = reader.read(&scope).await?;
700 let outcome = pull(&PullInput {
701 gateway: &gateway,
702 env,
703 target: &target,
704 manifest: &manifest,
705 observed: &observed,
706 })?;
707
708 write_pulled(
709 repo,
710 &gateway,
711 existing.as_ref(),
712 &outcome.source,
713 generated.as_deref(),
714 )?;
715
716 let removed = removed_unreadable_warnings(&gateway, env, existing.as_ref(), &outcome.skipped);
717 left_out.extend(outcome.skipped);
718 left_out.sort_by(|a, b| a.path.as_bytes().cmp(b.path.as_bytes()));
719 let mut warnings = outcome.rendered.warnings.clone();
720 warnings.extend(removed);
721 warnings.extend(observed.local_units.iter().map(|unit| {
722 format!(
723 "{unit} also exists in the `local` collection, which the gateway loads over the mode collection; pull did not read the `local` copy"
724 )
725 }));
726 Ok(DeployPullResult {
727 env: env.to_string(),
728 gateway: gateway.clone(),
729 mode: target.mode.clone(),
730 node: "master".to_string(),
731 transport: transport_label(&target.nodes.master).to_string(),
732 out: format!("gateways/{gateway}"),
733 files_written: outcome.source.files.len(),
734 manifest: if generated.is_some() {
735 "generated"
736 } else {
737 "kept"
738 }
739 .to_string(),
740 render_sha256: outcome.rendered.tree.render_sha256(),
741 skipped: left_out,
742 warnings,
743 excluded_tag_providers: excluded,
744 })
745}
746
747const REMOVED_FILES_SHOWN: usize = 10;
750
751fn removed_unreadable_warnings(
761 gateway: &str,
762 env: &str,
763 existing: Option<&Existing>,
764 skipped: &[Skipped],
765) -> Vec<String> {
766 let Some(existing) = existing else {
767 return Vec::new();
768 };
769 let mode_prefix = format!("modes/{env}/");
770 let mut out = Vec::new();
771 for unit in skipped.iter().filter(|s| s.reason == REASON_PATH_TOO_LONG) {
772 let under = format!("{}/", unit.path);
773 let removed: Vec<String> = existing
774 .files
775 .iter()
776 .filter(|file| {
777 let base = file.strip_prefix(&mode_prefix).unwrap_or(file);
778 RelPath::new(base)
779 .ok()
780 .and_then(|p| collection_path(&p).ok())
781 .is_some_and(|c| c.as_str() == unit.path || c.as_str().starts_with(&under))
782 })
783 .map(|file| format!("gateways/{gateway}/{file}"))
784 .collect();
785 if removed.is_empty() {
786 continue;
787 }
788 let take = removed.len().min(REMOVED_FILES_SHOWN);
789 let mut named = removed[..take].join(", ");
790 if removed.len() > take {
791 named.push_str(&format!(" and {} more", removed.len() - take));
792 }
793 out.push(format!(
794 "{} 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",
795 unit.path,
796 removed.len()
797 ));
798 }
799 out
800}
801
802fn safe_component(name: &str) -> bool {
805 !name.is_empty()
806 && name != "."
807 && name != ".."
808 && !name.contains(['/', '\\', '\0'])
809 && !name.chars().any(char::is_control)
810}
811
812fn validate_pulled(gateway: &str, source: &SourceTree) -> Result<(), CoreError> {
818 let refuse = |path: &str, why: String| {
819 CoreError::deploy_tree_invalid(format!("gateways/{gateway}/{path}"), None, None, why)
820 };
821 let mut folded: BTreeMap<String, &str> = BTreeMap::new();
822 let mut dir_folds: BTreeSet<String> = BTreeSet::new();
823 for path in source.files.keys() {
824 let parts: Vec<&str> = path.components().collect();
825 if parts.iter().any(|c| !safe_component(c)) {
826 return Err(refuse(
827 path.as_str(),
828 "has an unsafe path component; refused before anything was changed".to_string(),
829 ));
830 }
831 let fold = path.as_str().to_lowercase();
832 if let Some(other) = folded.insert(fold.clone(), path.as_str()) {
833 return Err(refuse(
834 path.as_str(),
835 format!(
836 "collides with {other} when case is folded; a case-insensitive filesystem keeps only one. Refused before anything was changed"
837 ),
838 ));
839 }
840 for end in fold.match_indices('/').map(|(i, _)| i) {
841 dir_folds.insert(fold[..end].to_string());
842 }
843 }
844 for (fold, path) in &folded {
845 if dir_folds.contains(fold) {
846 return Err(refuse(
847 path,
848 "is both a file and a directory in the pulled tree; refused before anything was changed"
849 .to_string(),
850 ));
851 }
852 }
853 Ok(())
854}
855
856fn write_pulled(
860 repo: &Path,
861 gateway: &str,
862 existing: Option<&Existing>,
863 source: &SourceTree,
864 manifest: Option<&str>,
865) -> Result<(), CoreError> {
866 validate_pulled(gateway, source)?;
867 write_pulled_fs(repo, gateway, existing, source, manifest)
868}
869
870#[cfg(unix)]
874fn write_pulled_fs(
875 repo: &Path,
876 gateway: &str,
877 existing: Option<&Existing>,
878 source: &SourceTree,
879 manifest: Option<&str>,
880) -> Result<(), CoreError> {
881 use crate::deploy::tree::dirfd::{self, FdError, Kind};
882 use std::io::Write;
883
884 let fail = |what: &str, e: FdError| match e {
885 FdError::Changed => CoreError::deploy_tree_invalid(
886 what,
887 None,
888 None,
889 "changed while it was written (a symlink or a non-directory appeared); refused, nothing outside the gateway directory was touched",
890 ),
891 FdError::Io(io) => CoreError::Internal(format!("cannot write {what}: {io}")),
892 other => CoreError::Internal(format!("cannot write {what}: {other:?}")),
893 };
894
895 let repo_real = std::fs::canonicalize(repo).map_err(|e| {
899 CoreError::Internal(format!("cannot resolve the repo {}: {e}", repo.display()))
900 })?;
901 let root = dirfd::open_root(&repo_real).map_err(|e| fail("the repo", e))?;
902 let gateways = dirfd::ensure_dir(&root, "gateways").map_err(|e| fail("gateways", e))?;
903 let gw_label = format!("gateways/{gateway}");
904 let gw = dirfd::ensure_dir(&gateways, gateway).map_err(|e| fail(&gw_label, e))?;
905
906 if let Some(existing) = existing {
907 for file in &existing.files {
908 let parts: Vec<&str> = file.split('/').collect();
909 let (name, parents) = parts.split_last().ok_or_else(|| {
910 CoreError::Internal("an existing file has an empty path".to_string())
911 })?;
912 let label = format!("{gw_label}/{file}");
913 let dir = dirfd::walk(&gw, parents).map_err(|e| fail(&label, e))?;
914 match dirfd::kind_at(&dir, name).map_err(|e| fail(&label, e))? {
915 Kind::File => dirfd::unlink_at(&dir, name).map_err(|e| fail(&label, e))?,
916 Kind::Missing => {}
917 _ => return Err(fail(&label, FdError::Changed)),
918 }
919 }
920 for dir_path in existing.dirs.iter().rev() {
922 let parts: Vec<&str> = dir_path.split('/').collect();
923 let Some((name, parents)) = parts.split_last() else {
924 continue;
925 };
926 let label = format!("{gw_label}/{dir_path}");
927 let parent = dirfd::walk(&gw, parents).map_err(|e| fail(&label, e))?;
928 match dirfd::rmdir_at(&parent, name) {
929 Ok(()) | Err(FdError::Missing) => {}
930 Err(FdError::NotEmpty) => {
931 return Err(CoreError::deploy_tree_invalid(
932 label,
933 None,
934 None,
935 "is not empty after the pull removed its files (something wrote into it); nothing more was changed",
936 ));
937 }
938 Err(e) => return Err(fail(&label, e)),
939 }
940 }
941 }
942
943 if let Some(text) = manifest {
944 let label = format!("{gw_label}/manifest.yaml");
945 let mut file = dirfd::create_new(&gw, "manifest.yaml").map_err(|e| fail(&label, e))?;
946 file.write_all(text.as_bytes())
947 .map_err(|e| CoreError::Internal(format!("cannot write {label}: {e}")))?;
948 }
949 for (path, bytes) in &source.files {
950 let label = format!("{gw_label}/{path}");
951 let parts: Vec<&str> = path.components().collect();
952 if parts.iter().any(|c| !safe_component(c)) {
953 return Err(CoreError::deploy_tree_invalid(
954 label,
955 None,
956 None,
957 "has an unsafe path component; refused",
958 ));
959 }
960 let (name, parents) = parts
961 .split_last()
962 .ok_or_else(|| CoreError::Internal("a pulled file has an empty path".to_string()))?;
963 let dir = dirfd::ensure_walk(&gw, parents).map_err(|e| fail(&label, e))?;
964 let mut file = dirfd::create_new(&dir, name).map_err(|e| fail(&label, e))?;
965 file.write_all(bytes)
966 .map_err(|e| CoreError::Internal(format!("cannot write {label}: {e}")))?;
967 }
968 Ok(())
969}
970
971#[cfg(not(unix))]
975fn write_pulled_fs(
976 repo: &Path,
977 gateway: &str,
978 existing: Option<&Existing>,
979 source: &SourceTree,
980 manifest: Option<&str>,
981) -> Result<(), CoreError> {
982 use std::io::Write;
983
984 let io = |what: &Path, e: std::io::Error| {
985 CoreError::Internal(format!("cannot write {}: {e}", what.display()))
986 };
987 let gw = repo.join("gateways").join(gateway);
988 std::fs::create_dir_all(&gw).map_err(|e| io(&gw, e))?;
989 if let Some(existing) = existing {
990 for file in &existing.files {
991 let path = file.split('/').fold(gw.clone(), |acc, c| acc.join(c));
992 std::fs::remove_file(&path).map_err(|e| io(&path, e))?;
993 }
994 for dir_path in existing.dirs.iter().rev() {
995 let path = dir_path.split('/').fold(gw.clone(), |acc, c| acc.join(c));
996 std::fs::remove_dir(&path).map_err(|e| io(&path, e))?;
997 }
998 }
999 let create = |path: &Path, bytes: &[u8]| -> Result<(), CoreError> {
1000 if let Some(parent) = path.parent() {
1001 std::fs::create_dir_all(parent).map_err(|e| io(parent, e))?;
1002 }
1003 let mut file = std::fs::OpenOptions::new()
1004 .write(true)
1005 .create_new(true)
1006 .open(path)
1007 .map_err(|e| io(path, e))?;
1008 file.write_all(bytes).map_err(|e| io(path, e))
1009 };
1010 if let Some(text) = manifest {
1011 create(&gw.join("manifest.yaml"), text.as_bytes())?;
1012 }
1013 for (path, bytes) in &source.files {
1014 let parts: Vec<&str> = path.components().collect();
1015 if parts.iter().any(|c| !safe_component(c)) {
1016 return Err(CoreError::deploy_tree_invalid(
1017 format!("gateways/{gateway}/{path}"),
1018 None,
1019 None,
1020 "has an unsafe path component; refused",
1021 ));
1022 }
1023 let dest = parts.iter().fold(gw.clone(), |acc, c| acc.join(c));
1024 create(&dest, bytes)?;
1025 }
1026 Ok(())
1027}
1028
1029#[derive(Debug, Clone)]
1031pub struct VerifyOptions {
1032 pub probe_deadline: Duration,
1034 pub probe_interval: Duration,
1036 pub log_settle: Duration,
1039}
1040
1041impl Default for VerifyOptions {
1042 fn default() -> Self {
1043 Self {
1044 probe_deadline: verify::PROBE_DEADLINE,
1045 probe_interval: verify::PROBE_INTERVAL,
1046 log_settle: Duration::from_millis(verify::LOG_SETTLE_MS.unsigned_abs()),
1047 }
1048 }
1049}
1050
1051#[derive(Debug, Clone, PartialEq, Serialize)]
1053pub struct DeployVerifyResult {
1054 pub env: String,
1056 pub gateway: String,
1058 pub mode: String,
1060 pub node: String,
1063 pub project: String,
1065 pub ok: bool,
1067 pub slug: Option<&'static str>,
1069 pub checks: Vec<verify::Check>,
1071 pub resources_checked: usize,
1073 pub warnings: Vec<String>,
1075 pub excluded_tag_providers: Vec<String>,
1078 #[serde(skip_serializing_if = "Option::is_none")]
1081 pub backup: Option<BackupReport>,
1082 #[serde(skip_serializing_if = "Option::is_none")]
1085 pub secrets: Option<SecretNamesReport>,
1086}
1087
1088impl DeployVerifyResult {
1089 pub fn failure_exit_code(&self) -> Option<u8> {
1094 (!self.ok).then_some(6)
1095 }
1096}
1097
1098const LOG_PAGES_MAX: usize = 5;
1100
1101pub async fn deploy_verify(
1124 declared: &Declared,
1125 project: &str,
1126 deploy: &dyn DeployApi,
1127 gateway: &dyn GatewayApi,
1128 reader: &dyn TreeReader,
1129 opts: &VerifyOptions,
1130) -> Result<DeployVerifyResult, CoreError> {
1131 deploy_verify_judged(
1132 declared,
1133 project,
1134 deploy,
1135 gateway,
1136 reader,
1137 opts,
1138 VerifyContext::Standalone,
1139 )
1140 .await
1141}
1142
1143#[derive(Debug, Clone, Copy)]
1146pub enum VerifyContext<'a> {
1147 Standalone,
1149 PostApply(&'a verify::ReadBackBaseline),
1152}
1153
1154pub(crate) async fn deploy_verify_judged(
1156 declared: &Declared,
1157 project: &str,
1158 deploy: &dyn DeployApi,
1159 gateway: &dyn GatewayApi,
1160 reader: &dyn TreeReader,
1161 opts: &VerifyOptions,
1162 ctx: VerifyContext<'_>,
1163) -> Result<DeployVerifyResult, CoreError> {
1164 deploy_verify_planned(declared, project, deploy, gateway, reader, opts, ctx)
1165 .await
1166 .map(|(result, _)| result)
1167}
1168
1169async fn deploy_verify_planned(
1174 declared: &Declared,
1175 project: &str,
1176 deploy: &dyn DeployApi,
1177 gateway: &dyn GatewayApi,
1178 reader: &dyn TreeReader,
1179 opts: &VerifyOptions,
1180 ctx: VerifyContext<'_>,
1181) -> Result<(DeployVerifyResult, BTreeMap<String, PlanStatus>), CoreError> {
1182 let expected =
1183 verify::Expected::from_render(&declared.target.mode, &declared.outcome, &declared.manifest);
1184
1185 let scan = deploy.scan_status(ScanKind::Config).await?;
1186 let scan_read_at = std::time::Instant::now();
1187 let mode = normalize_deployment_mode(deploy.deployment_mode().await?);
1188 let read_back = gather_read_back(deploy, &expected).await?;
1189
1190 let excluded = declared.excluded_tag_providers();
1193 let scope = Scope::from_manifest(&declared.manifest, &declared.target.mode, &excluded);
1194 let live = reader.read(&scope).await?;
1195 let planned = plan(&declared.outcome.tree, &live, None, &scope, &excluded);
1196 let files = planned
1197 .entries
1198 .iter()
1199 .filter(|entry| expected.tag_units.contains(&entry.key))
1200 .map(|entry| (entry.key.clone(), entry.status))
1201 .collect();
1202 let file_status = resource_statuses(&expected, &planned.entries);
1203
1204 let logs = if scan.scan_active {
1205 Ok(Vec::new())
1207 } else {
1208 settle_logs(gateway, &scan, scan_read_at, opts.log_settle).await?;
1209 read_logs(gateway, verify::log_window_start(&scan)).await?
1210 };
1211 let probes = if expected.probes.is_empty() {
1212 None
1213 } else {
1214 Some(poll_probes(gateway, project, &expected.probes, opts).await?)
1215 };
1216
1217 let observed = verify::ObservedState {
1218 scan,
1219 mode,
1220 read_back,
1221 files,
1222 logs,
1223 probes,
1224 unreadable: planned
1225 .unreadable
1226 .iter()
1227 .map(|unit| (unit.key.clone(), unit.declared))
1228 .collect(),
1229 };
1230 let judge = match ctx {
1231 VerifyContext::Standalone => verify::ReadBackJudge::Standalone {
1232 file_status: &file_status,
1233 },
1234 VerifyContext::PostApply(baseline) => verify::ReadBackJudge::Baseline(baseline),
1235 };
1236 let report = verify::evaluate_with(&expected, &observed, &judge);
1237 let result = DeployVerifyResult {
1238 env: declared.env.clone(),
1239 gateway: declared.target.gateway.clone(),
1240 mode: declared.target.mode.clone(),
1241 node: "master".to_string(),
1242 project: project.to_string(),
1243 ok: report.ok,
1244 slug: report.slug,
1245 checks: report.checks,
1246 resources_checked: report.resources_checked,
1247 warnings: report.warnings,
1248 excluded_tag_providers: excluded,
1249 backup: None,
1250 secrets: None,
1251 };
1252 Ok((result, file_status))
1253}
1254
1255fn resource_statuses(
1258 expected: &verify::Expected,
1259 entries: &[PlanEntry],
1260) -> BTreeMap<String, PlanStatus> {
1261 let keys: BTreeSet<&str> = expected.resources.iter().map(|r| r.key.as_str()).collect();
1262 entries
1263 .iter()
1264 .filter(|entry| keys.contains(entry.key.as_str()))
1265 .map(|entry| (entry.key.clone(), entry.status))
1266 .collect()
1267}
1268
1269#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1271pub enum VerifyNode {
1272 Master,
1275 Backup,
1277 Both,
1279}
1280
1281impl VerifyNode {
1282 pub fn as_str(self) -> &'static str {
1284 match self {
1285 VerifyNode::Master => "master",
1286 VerifyNode::Backup => "backup",
1287 VerifyNode::Both => "both",
1288 }
1289 }
1290}
1291
1292pub type VerifyMasterSurfaces<'a> = (&'a dyn DeployApi, &'a dyn GatewayApi);
1294
1295pub async fn deploy_verify_node(
1317 declared: &Declared,
1318 project: &str,
1319 node: VerifyNode,
1320 master: Option<VerifyMasterSurfaces<'_>>,
1321 master_reader: &dyn TreeReader,
1322 backup: Option<&BackupNode<'_>>,
1323 opts: &VerifyOptions,
1324) -> Result<DeployVerifyResult, CoreError> {
1325 let missing = |what: &str| CoreError::InvalidInput {
1326 reason: format!("deploy verify --node {} needs the {what}", node.as_str()),
1327 };
1328 let (master_result, master_status) = if node == VerifyNode::Backup {
1329 (None, None)
1330 } else {
1331 let (deploy, gateway) = master.ok_or_else(|| missing("master surfaces"))?;
1332 let (result, status) = deploy_verify_planned(
1333 declared,
1334 project,
1335 deploy,
1336 gateway,
1337 master_reader,
1338 opts,
1339 VerifyContext::Standalone,
1340 )
1341 .await?;
1342 (Some(result), Some(status))
1343 };
1344 if node == VerifyNode::Master {
1345 return master_result.ok_or_else(|| missing("master surfaces"));
1346 }
1347 let backup = backup.ok_or_else(|| missing("backup node"))?;
1348 let expected =
1349 verify::Expected::from_render(&declared.target.mode, &declared.outcome, &declared.manifest);
1350 let mut backup_warnings = Vec::new();
1351 let (tick, compared) = verify_backup_tick(
1352 declared,
1353 &expected,
1354 master_reader,
1355 backup,
1356 master_status.as_ref(),
1357 &mut backup_warnings,
1358 )
1359 .await;
1360 let report = verify_report(&tick);
1361
1362 let mut result = match master_result {
1363 Some(mut result) => {
1367 result.warnings.extend(backup_warnings);
1368 result
1369 }
1370 None => DeployVerifyResult {
1371 env: declared.env.clone(),
1372 gateway: declared.target.gateway.clone(),
1373 mode: declared.target.mode.clone(),
1374 node: String::new(),
1375 project: project.to_string(),
1376 ok: true,
1377 slug: None,
1378 checks: Vec::new(),
1379 resources_checked: compared,
1380 warnings: backup_warnings,
1381 excluded_tag_providers: declared.excluded_tag_providers(),
1382 backup: None,
1383 secrets: None,
1384 },
1385 };
1386 let secrets = verify_secret_names(
1389 declared,
1390 master
1391 .map(|(deploy, _)| deploy)
1392 .filter(|_| node != VerifyNode::Backup),
1393 Some(backup),
1394 )
1395 .await?;
1396 let secrets_ok = secrets.nodes.iter().all(|n| n.status != CheckStatus::Fail);
1397 result.node = node.as_str().to_string();
1398 result.ok = result.ok && report.converged && secrets_ok;
1399 result.slug = (!result.ok).then_some(verify::VERIFY_FAILED_SLUG);
1400 result.backup = Some(report);
1401 result.secrets = Some(secrets);
1402 Ok(result)
1403}
1404
1405async fn verify_secret_names(
1412 declared: &Declared,
1413 master: Option<&dyn DeployApi>,
1414 backup: Option<&BackupNode<'_>>,
1415) -> Result<SecretNamesReport, CoreError> {
1416 let referenced = collect_referenced_secrets(&declared.outcome.tree);
1417 let declared_providers = declared_secret_providers(&declared.outcome.tree);
1418 let providers: BTreeSet<&str> = referenced.iter().map(|(p, _)| p.as_str()).collect();
1419 let mut nodes = Vec::with_capacity(2);
1420 if let Some(master) = master {
1421 let mut listings = BTreeMap::new();
1422 for provider in &providers {
1423 listings.insert(
1424 (*provider).to_string(),
1425 master.secret_names(provider).await?,
1426 );
1427 }
1428 nodes.push(check_secret_names(
1429 "master",
1430 &referenced,
1431 &listings,
1432 &declared_providers,
1433 SecretCheckMode::Verify,
1434 ));
1435 }
1436 if let Some(backup) = backup {
1437 let mut listings = BTreeMap::new();
1438 for provider in &providers {
1439 listings.insert(
1440 (*provider).to_string(),
1441 backup.secret_names(provider).await?,
1442 );
1443 }
1444 nodes.push(check_secret_names(
1445 "backup",
1446 &referenced,
1447 &listings,
1448 &declared_providers,
1449 SecretCheckMode::Verify,
1450 ));
1451 }
1452 Ok(SecretNamesReport::ordered(nodes))
1453}
1454
1455async fn verify_backup_tick(
1468 declared: &Declared,
1469 expected: &verify::Expected,
1470 master_reader: &dyn TreeReader,
1471 backup: &BackupNode<'_>,
1472 master_status: Option<&BTreeMap<String, PlanStatus>>,
1473 warnings: &mut Vec<String>,
1474) -> (TickOutcome, usize) {
1475 let signals = match backup.status().await {
1476 Ok(status) => Ok(BackupSignals::from_status(None, &status)),
1477 Err(error) => Err(error.code().to_string()),
1478 };
1479 let mode = match backup.deployment_mode().await {
1480 Ok(mode) => mode_backup_check(expected, mode.as_deref()),
1481 Err(error) => unreachable_check(BackupCheckId::Mode, error.code()),
1482 };
1483 let mut files_warnings = Vec::new();
1484 let (files, files_checked, files_status) = verify_backup_files(
1485 declared,
1486 expected,
1487 master_reader,
1488 backup,
1489 master_status.is_none(),
1490 &mut files_warnings,
1491 )
1492 .await;
1493 let file_status = master_status.or(files_status.as_ref());
1494 let judge = match file_status {
1495 Some(file_status) => verify::ReadBackJudge::Standalone { file_status },
1496 None => verify::ReadBackJudge::Strict,
1497 };
1498 let mut compared = 0;
1499 let read_back = match backup.read_back(expected).await {
1500 Ok(read_back) => {
1501 let (check, count) =
1502 read_back_backup_check_with(expected, &read_back, &judge, warnings);
1503 compared = count;
1504 check
1505 }
1506 Err(error) => unreachable_check(BackupCheckId::ReadBack, error.code()),
1507 };
1508 warnings.extend(files_warnings);
1509 (
1510 TickOutcome {
1511 signals,
1512 exhausted: false,
1513 content: Some(vec![mode, read_back, files]),
1514 files_checked,
1515 stable_ticks: 0,
1516 },
1517 compared,
1518 )
1519}
1520
1521async fn verify_backup_files(
1530 declared: &Declared,
1531 expected: &verify::Expected,
1532 master_reader: &dyn TreeReader,
1533 backup: &BackupNode<'_>,
1534 want_status: bool,
1535 warnings: &mut Vec<String>,
1536) -> (BackupCheck, usize, Option<BTreeMap<String, PlanStatus>>) {
1537 let scope = Scope::from_manifest(
1538 &declared.manifest,
1539 &declared.target.mode,
1540 &declared.excluded_tag_providers(),
1541 );
1542 let backup_files = match backup.read_files(&scope).await {
1543 None => return (files_skipped_check(), 0, None),
1545 Some(Ok(observed)) => observed,
1546 Some(Err(error)) => {
1547 return (
1548 unreachable_check(BackupCheckId::Files, &format!("backup {}", error.code())),
1549 0,
1550 None,
1551 );
1552 }
1553 };
1554 let backup_unreadable = unreadable_units(&backup_files);
1555 let backup_files = effective_files(backup_files);
1556 let master_files = match master_reader.read(&scope).await {
1557 Ok(observed) => observed,
1558 Err(error) => {
1559 return (
1560 unreachable_check(BackupCheckId::Files, &format!("master {}", error.code())),
1561 0,
1562 None,
1563 );
1564 }
1565 };
1566 let status = want_status.then(|| {
1567 let excluded = declared.excluded_tag_providers();
1568 let planned = plan(
1569 &declared.outcome.tree,
1570 &master_files,
1571 None,
1572 &scope,
1573 &excluded,
1574 );
1575 resource_statuses(expected, &planned.entries)
1576 });
1577 let master_unreadable = unreadable_units(&master_files);
1578 let master_files = effective_files(master_files);
1579 let present: BTreeSet<_> = declared.outcome.tree.files.keys().cloned().collect();
1580 let absent = BTreeSet::new();
1581 let mut comparison = compare_files(&master_files, &backup_files, &present, &absent);
1582 exempt_unreadable(
1583 &mut comparison,
1584 &present,
1585 &absent,
1586 &[
1587 ("master", &master_unreadable),
1588 ("backup", &backup_unreadable),
1589 ],
1590 );
1591 warnings.extend(unreadable_warnings(&comparison));
1592 (files_check(&comparison), comparison.checked, status)
1593}
1594
1595pub(crate) async fn gather_read_back(
1600 deploy: &dyn DeployApi,
1601 expected: &verify::Expected,
1602) -> Result<BTreeMap<String, verify::Remote>, CoreError> {
1603 let classify = |record: crate::client::config_resources::ResourceRecord| {
1604 if record.collection.as_deref() == Some(verify::SYSTEM_COLLECTION) {
1605 verify::Remote::System
1606 } else {
1607 verify::Remote::Config(record.config)
1608 }
1609 };
1610 let mut lists: BTreeMap<(String, String), Option<Vec<_>>> = BTreeMap::new();
1611 let mut out = BTreeMap::new();
1612 for res in &expected.resources {
1613 let remote = match &res.name {
1614 None => match deploy.resource_singleton(&res.module, &res.ty, None).await {
1615 Ok(record) => classify(record),
1616 Err(CoreError::NotFound { .. }) => verify::Remote::Missing,
1617 Err(e) => return Err(e),
1618 },
1619 Some(name) => {
1620 let type_key = (res.module.clone(), res.ty.clone());
1621 if !lists.contains_key(&type_key) {
1622 let items = match deploy
1623 .resource_list(&res.module, &res.ty, &ListQuery::default())
1624 .await
1625 {
1626 Ok(envelope) => Some(envelope.items),
1627 Err(CoreError::NotFound { .. }) => None,
1628 Err(e) => return Err(e),
1629 };
1630 lists.insert(type_key.clone(), items);
1631 }
1632 lists
1633 .get(&type_key)
1634 .and_then(Option::as_ref)
1635 .and_then(|items| items.iter().find(|r| r.name.as_deref() == Some(name)))
1636 .map_or(verify::Remote::Missing, |r| classify(r.clone()))
1637 }
1638 };
1639 out.insert(res.key.clone(), remote);
1640 }
1641 Ok(out)
1642}
1643
1644async fn settle_logs(
1648 gateway: &dyn GatewayApi,
1649 scan: &ScanStatus,
1650 scan_read_at: std::time::Instant,
1651 settle: Duration,
1652) -> Result<(), CoreError> {
1653 if settle.is_zero() {
1654 return Ok(());
1655 }
1656 let settle_ms = i64::try_from(settle.as_millis()).unwrap_or(i64::MAX);
1657 let probe = LogQuery {
1658 start_time: Some(scan.last_scan_timestamp.saturating_add(settle_ms)),
1659 limit: 1,
1660 ..LogQuery::default()
1661 };
1662 match gateway.logs(&probe).await {
1663 Ok(envelope) if !envelope.items.is_empty() => return Ok(()),
1664 Err(e @ CoreError::Network { .. }) => return Err(e),
1665 Ok(_) | Err(_) => {}
1667 }
1668 tokio::time::sleep(settle.saturating_sub(scan_read_at.elapsed())).await;
1669 Ok(())
1670}
1671
1672async fn read_logs(
1676 gateway: &dyn GatewayApi,
1677 start_time: i64,
1678) -> Result<Result<Vec<verify::LogLine>, String>, CoreError> {
1679 let mut lines = Vec::new();
1680 let mut offset = 0i64;
1681 for _ in 0..LOG_PAGES_MAX {
1682 let query = LogQuery {
1683 start_time: Some(start_time),
1684 min_level: Some("WARN".to_string()),
1685 limit: DEFAULT_LOG_LIMIT,
1686 offset,
1687 ..LogQuery::default()
1688 };
1689 match gateway.logs(&query).await {
1690 Ok(envelope) => {
1691 let count = i64::try_from(envelope.items.len()).unwrap_or(i64::MAX);
1692 lines.extend(envelope.items.into_iter().map(|e| verify::LogLine {
1693 timestamp: e.timestamp,
1694 logger: e.logger_name,
1695 message: e.message,
1696 }));
1697 if count < DEFAULT_LOG_LIMIT {
1698 return Ok(Ok(lines));
1699 }
1700 offset += count;
1701 }
1702 Err(e @ CoreError::Network { .. }) => return Err(e),
1703 Err(e) => return Ok(Err(e.code().to_string())),
1704 }
1705 }
1706 Ok(Err("logs_truncated".to_string()))
1707}
1708
1709async fn poll_probes(
1714 gateway: &dyn GatewayApi,
1715 project: &str,
1716 paths: &[String],
1717 opts: &VerifyOptions,
1718) -> Result<Result<Vec<verify::ProbeReading>, String>, CoreError> {
1719 let interval = opts.probe_interval.max(Duration::from_millis(1));
1720 let started = tokio::time::Instant::now();
1721 loop {
1722 let readings: Vec<verify::ProbeReading> =
1723 match crate::actions::tags::tags_read(gateway, project, paths).await {
1724 Ok(result) => result
1725 .results
1726 .into_iter()
1727 .enumerate()
1728 .map(|(i, row)| verify::ProbeReading {
1729 path: if row.path.is_empty() {
1731 paths.get(i).cloned().unwrap_or_default()
1732 } else {
1733 row.path
1734 },
1735 quality: row.quality,
1736 })
1737 .collect(),
1738 Err(e @ CoreError::Network { .. }) => return Err(e),
1739 Err(e) => return Ok(Err(e.code().to_string())),
1740 };
1741 let all_good = paths.iter().all(|path| {
1742 readings
1743 .iter()
1744 .any(|r| &r.path == path && r.quality.starts_with("Good"))
1745 });
1746 if all_good || started.elapsed() + interval > opts.probe_deadline {
1747 return Ok(Ok(readings));
1748 }
1749 tokio::time::sleep(interval).await;
1750 }
1751}
1752
1753#[cfg(test)]
1754mod tests {
1755 use super::*;
1756
1757 fn repo_with_existing() -> (tempfile::TempDir, Existing) {
1759 let repo = tempfile::TempDir::new().expect("repo");
1760 let dir = repo.path().join("gateways/g/resources/ignition/x");
1761 std::fs::create_dir_all(&dir).expect("dirs");
1762 std::fs::write(dir.join("keep.json"), b"{\"keep\": true}").expect("file");
1763 let existing = Existing {
1764 files: vec!["resources/ignition/x/keep.json".to_string()],
1765 dirs: ["resources", "resources/ignition", "resources/ignition/x"]
1766 .iter()
1767 .map(|d| d.to_string())
1768 .collect(),
1769 };
1770 (repo, existing)
1771 }
1772
1773 fn snapshot(root: &Path) -> BTreeMap<PathBuf, Vec<u8>> {
1774 fn walk(dir: &Path, out: &mut BTreeMap<PathBuf, Vec<u8>>) {
1775 for entry in std::fs::read_dir(dir).expect("read_dir") {
1776 let path = entry.expect("entry").path();
1777 if path.is_dir() {
1778 out.insert(path.clone(), Vec::new());
1779 walk(&path, out);
1780 } else {
1781 out.insert(path.clone(), std::fs::read(&path).expect("read"));
1782 }
1783 }
1784 }
1785 let mut out = BTreeMap::new();
1786 walk(root, &mut out);
1787 out
1788 }
1789
1790 fn assert_refused_untouched(files: &[(&str, &str)], needle: &str) {
1793 let (repo, existing) = repo_with_existing();
1794 let before = snapshot(repo.path());
1795 let source = SourceTree::from_files(files.iter().copied()).expect("source");
1796 let err =
1797 write_pulled(repo.path(), "g", Some(&existing), &source, None).expect_err("refused");
1798 assert_eq!(err.code(), "deploy_tree_invalid", "{err}");
1799 assert!(err.to_string().contains(needle), "{needle}: {err}");
1800 assert_eq!(
1801 snapshot(repo.path()),
1802 before,
1803 "a refused pull must not delete the existing tree"
1804 );
1805 }
1806
1807 #[test]
1808 fn unsafe_component_is_refused_before_any_delete() {
1809 assert_refused_untouched(
1810 &[
1811 ("resources/ignition/x/fine.json", "{}"),
1812 ("resources/ignition/x/bad\nname.json", "{}"),
1813 ],
1814 "unsafe path component",
1815 );
1816 }
1817
1818 #[test]
1819 fn case_fold_collision_is_refused_before_any_delete() {
1820 assert_refused_untouched(
1821 &[
1822 ("resources/ignition/x/Foo.json", "{}"),
1823 ("resources/ignition/x/foo.json", "{}"),
1824 ],
1825 "case is folded",
1826 );
1827 }
1828
1829 #[test]
1830 fn file_and_directory_clash_is_refused_before_any_delete() {
1831 assert_refused_untouched(
1832 &[
1833 ("resources/ignition/x", "{}"),
1834 ("resources/ignition/x/a.json", "{}"),
1835 ],
1836 "both a file and a directory",
1837 );
1838 }
1839
1840 #[test]
1841 fn a_valid_tree_replaces_the_existing_one() {
1842 let (repo, existing) = repo_with_existing();
1843 let source =
1844 SourceTree::from_files([("resources/ignition/x/new.json", "{}")]).expect("source");
1845 write_pulled(repo.path(), "g", Some(&existing), &source, None).expect("written");
1846 let dir = repo.path().join("gateways/g/resources/ignition/x");
1847 assert!(!dir.join("keep.json").exists());
1848 assert!(dir.join("new.json").is_file());
1849 }
1850}