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::{PAIR_UNSYNCED_TAG_TYPES, 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_mode_tag_dirs, pair_mode_tag_files, pair_tag_keys, plan_with,
108 render_diffs,
109};
110use crate::deploy::pull::{PullInput, REASON_PATH_TOO_LONG, Skipped, generate_manifest, pull};
111use crate::deploy::render::{RenderInput, RenderOutcome, RenderedResource, render};
112use crate::deploy::target::{NodeTarget, Redundancy, Target};
113use crate::deploy::transport::{CORE_COLLECTION, DeleteSet, READ_MAX_ENTRIES, Scope, TreeReader};
114use crate::deploy::tree::{
115 RenderMarker, SourceTree, read_repo_file, require_repo_dir, walk_capped, write_out,
116};
117use crate::deploy::verify::{self, CheckStatus};
118use crate::error::CoreError;
119
120#[derive(Debug, Clone, PartialEq, Serialize)]
122pub struct DeployRenderResult {
123 pub env: String,
125 pub gateway: String,
127 pub mode: String,
129 pub out: Option<String>,
131 pub files: usize,
133 pub bytes: usize,
135 pub render_sha256: String,
137 pub resources: Vec<RenderedResource>,
139 pub tag_providers: Vec<String>,
141 pub uuids_minted: Vec<String>,
143 pub opaque_files: Vec<String>,
145 pub warnings: Vec<String>,
147}
148
149pub fn load_target(repo: &Path, env: &str) -> Result<(String, Target), CoreError> {
152 if let Some(problem) = component_problem(env) {
153 return Err(CoreError::InvalidInput {
154 reason: format!("--env {env}: {problem}"),
155 });
156 }
157 let target_rel = format!("targets/{env}.yaml");
158 let target_bytes = read_repo_file(
159 repo,
160 &target_rel,
161 Some(&format!("--env {env} selects {target_rel}")),
162 )?;
163 let target = Target::parse(&target_rel, &target_bytes)?;
164 Ok((target_rel, target))
165}
166
167#[derive(Debug, Clone)]
169pub struct Declared {
170 pub env: String,
172 pub target_file: String,
174 pub target: Target,
176 pub manifest: Manifest,
178 pub outcome: RenderOutcome,
180}
181
182impl Declared {
183 pub fn excluded_tag_providers(&self) -> Vec<String> {
187 exclude::effective(&self.target, Some(&self.manifest))
188 }
189}
190
191pub fn load_declared(repo: &Path, env: &str) -> Result<Declared, CoreError> {
194 load_declared_in(repo, repo, env)
195}
196
197pub fn load_declared_in(repo: &Path, targets: &Path, env: &str) -> Result<Declared, CoreError> {
203 let (target_file, target) = load_target(targets, env)?;
204
205 let gateway_rel = format!("gateways/{}", target.gateway);
206 let gateway_dir = repo.join("gateways").join(&target.gateway);
207 require_repo_dir(repo, "gateways")?;
208 require_repo_dir(repo, &gateway_rel)?;
209
210 let manifest_rel = format!("{gateway_rel}/manifest.yaml");
211 let manifest_bytes = read_repo_file(repo, &manifest_rel, None)?;
212 let manifest = Manifest::parse(&manifest_rel, &manifest_bytes)?;
213
214 let source = SourceTree::read_gateway(&gateway_dir, env)?;
215 let outcome = render(&RenderInput {
216 gateway: &target.gateway,
217 env,
218 manifest: &manifest,
219 target: &target,
220 source: &source,
221 })?;
222 Ok(Declared {
223 env: env.to_string(),
224 target_file,
225 target,
226 manifest,
227 outcome,
228 })
229}
230
231pub fn deploy_render(
234 repo: &Path,
235 env: &str,
236 out: Option<&Path>,
237) -> Result<DeployRenderResult, CoreError> {
238 deploy_render_in(repo, repo, env, out)
239}
240
241pub fn deploy_render_in(
245 repo: &Path,
246 targets: &Path,
247 env: &str,
248 out: Option<&Path>,
249) -> Result<DeployRenderResult, CoreError> {
250 let Declared {
251 target, outcome, ..
252 } = load_declared_in(repo, targets, env)?;
253
254 if let Some(out) = out {
256 let marker = RenderMarker::for_tree(&target.gateway, env, &target.mode, &outcome.tree);
257 write_out(&outcome.tree, out, &marker)?;
258 }
259
260 Ok(DeployRenderResult {
261 env: env.to_string(),
262 gateway: target.gateway.clone(),
263 mode: target.mode.clone(),
264 out: out.map(|p| p.to_string_lossy().into_owned()),
265 files: outcome.tree.files.len(),
266 bytes: outcome.tree.bytes(),
267 render_sha256: outcome.tree.render_sha256(),
268 resources: outcome.resources,
269 tag_providers: outcome.tag_providers,
270 uuids_minted: outcome.uuids_minted,
271 opaque_files: outcome.opaque_files,
272 warnings: outcome.warnings,
273 })
274}
275
276#[derive(Debug, Clone, PartialEq, Serialize)]
278pub struct DeployPlanResult {
279 pub env: String,
281 pub gateway: String,
283 pub mode: String,
285 pub node: String,
287 pub transport: String,
289 pub render_sha256: String,
291 pub plan_sha256: String,
293 pub summary: PlanSummary,
295 pub entries: Vec<PlanEntry>,
297 pub delete_set: DeleteSet,
300 pub base: Option<BaseSummary>,
302 pub warnings: Vec<String>,
304 pub excluded_tag_providers: Vec<String>,
308 pub unreadable: Vec<UnreadableUnit>,
313 pub provider_wipes: Vec<String>,
321 #[serde(skip)]
330 pub pair_mode_tag_dirs: Vec<String>,
331}
332
333#[derive(Debug, Clone, Default)]
335pub struct PlanOptions {
336 pub base: Option<PathBuf>,
339 pub diff: bool,
341}
342
343#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
345pub struct BaseSummary {
346 pub dir: String,
348 pub render_sha256: String,
350 pub files: usize,
352}
353
354pub async fn deploy_plan(
358 declared: &Declared,
359 reader: &dyn TreeReader,
360 opts: &PlanOptions,
361) -> Result<DeployPlanResult, CoreError> {
362 let node = &declared.target.nodes.master;
363 let transport = match (&node.ssh, &node.local) {
364 (Some(_), _) => "ssh",
365 (None, Some(_)) => "local",
366 (None, None) => "none",
367 };
368 let base = match &opts.base {
371 Some(dir) => {
372 let dir = dir.clone();
373 let (gateway, env, mode) = (
374 declared.target.gateway.clone(),
375 declared.env.clone(),
376 declared.target.mode.clone(),
377 );
378 let read =
379 tokio::task::spawn_blocking(move || BaseTree::read(&dir, &gateway, &env, &mode))
380 .await
381 .map_err(|e| CoreError::Internal(format!("--base read task failed: {e}")))?;
382 Some(read?)
383 }
384 None => None,
385 };
386 if let Some(journal) = journal_on_box(
391 reader,
392 node,
393 &declared.target.gateway,
394 &declared.target.mode,
395 )
396 .await?
397 {
398 return Err(journal_present(
399 &journal,
400 &declared.env,
401 node,
402 &declared.target.gateway,
403 &declared.target.mode,
404 ));
405 }
406 let excluded = declared.excluded_tag_providers();
407 let scope = Scope::from_manifest(&declared.manifest, &declared.target.mode, &excluded);
408 let observed = reader.read(&scope).await?;
409 let clashes = live_tag_folder_findings(&observed, &declared.target.mode);
413 if !clashes.is_empty() {
414 return Err(CoreError::deploy_findings(clashes));
415 }
416 let gateway = &declared.target.gateway;
417 let mode = &declared.target.mode;
418 let ledger = match reader.read_ledger(gateway, mode).await? {
421 Some(bytes) => Some(Ledger::parse(
422 &ledger_location(LedgerRoot::for_node(node), gateway, mode),
423 &bytes,
424 gateway,
425 mode,
426 )?),
427 None => None,
428 };
429 let mut planned = plan_with(
432 &declared.outcome.tree,
433 &observed,
434 ledger.as_ref(),
435 &scope,
436 &excluded,
437 Redundancy::of(&declared.target),
438 );
439 if let Some(base) = &base {
440 apply_base(&mut planned, base);
441 }
442 if opts.diff {
443 render_diffs(&mut planned, &declared.outcome.tree, &observed);
444 }
445
446 let mut warnings = declared.outcome.warnings.clone();
447 warnings.extend(planned.warnings);
448 let mut pair_mode_tag_dirs = Vec::new();
449 if declared.target.nodes.backup.is_some() {
450 pair_mode_tag_dirs = pair_mode_tag_dirs_of(reader, mode, &planned.entries).await?;
451 warnings.extend(pair_tags_warning(
452 &planned.entries,
453 &planned.delete_set,
454 &pair_mode_tag_dirs,
455 mode,
456 ));
457 }
458 Ok(DeployPlanResult {
459 env: declared.env.clone(),
460 gateway: declared.target.gateway.clone(),
461 mode: declared.target.mode.clone(),
462 node: "master".to_string(),
463 transport: transport.to_string(),
464 render_sha256: declared.outcome.tree.render_sha256(),
465 plan_sha256: planned.plan_sha256,
466 summary: planned.summary,
467 entries: planned.entries,
468 delete_set: planned.delete_set,
469 base: base.map(|b| BaseSummary {
470 dir: b.dir,
471 render_sha256: b.render_sha256,
472 files: b.files.len(),
473 }),
474 warnings,
475 excluded_tag_providers: excluded,
476 unreadable: planned.unreadable,
477 provider_wipes: planned.provider_wipes,
478 pair_mode_tag_dirs,
479 })
480}
481
482pub(crate) async fn pair_mode_tag_dirs_of(
490 reader: &dyn TreeReader,
491 mode: &str,
492 entries: &[PlanEntry],
493) -> Result<Vec<String>, CoreError> {
494 let dirs: Vec<(String, String)> = PAIR_UNSYNCED_TAG_TYPES
495 .iter()
496 .map(|ty| ("ignition".to_string(), (*ty).to_string()))
497 .collect();
498 let listed = reader.list_tree(mode, &dirs).await?;
499 Ok(pair_mode_tag_dirs(&listed, entries))
500}
501
502fn named_list(items: &[String]) -> String {
504 let shown = items.len().min(crate::error::APPLY_LIST_SHOWN);
505 let mut named = items[..shown].join(", ");
506 if items.len() > shown {
507 named.push_str(&format!(" and {} more", items.len() - shown));
508 }
509 named
510}
511
512fn pair_tags_warning(
524 entries: &[PlanEntry],
525 delete: &DeleteSet,
526 dirs: &[String],
527 mode: &str,
528) -> Option<String> {
529 let mut keys = pair_tag_keys(entries, delete);
530 keys.extend(dirs.iter().cloned());
531 keys.sort_by(|a, b| a.as_bytes().cmp(b.as_bytes()));
532 keys.dedup();
533 if keys.is_empty() {
534 return None;
535 }
536 Some(format!(
537 "redundant pair: {} tag unit(s) or folder(s) would leave files or folders in the {mode} \
538 collection that this deploy cannot delete (the ledger does not list them, the delete \
539 set skips them, or they are files or folders the deploy does not read), which Ignition's \
540 redundancy sync cannot carry; deploy apply refuses this plan \
541 (deploy_pair_tags_refused): {}",
542 keys.len(),
543 named_list(&keys)
544 ))
545}
546
547fn pair_mode_tags_warning(files: Vec<String>, dirs: &[String], mode: &str) -> Option<String> {
560 let mut paths = files;
561 paths.extend(dirs.iter().cloned());
562 if paths.is_empty() {
563 return None;
564 }
565 Some(format!(
566 "redundant pair: the master's {mode} collection holds {} tag file(s) or folder(s), which \
567 Ignition's redundancy sync cannot carry (the backup stops syncing configuration); a \
568 deploy that declares the units and lists the files in its ledger moves them to the core \
569 collection, or remove them on the master (README runbook T6e): {}",
570 paths.len(),
571 named_list(&paths)
572 ))
573}
574
575#[derive(Debug, Clone, PartialEq, Serialize)]
577pub struct DeployPullResult {
578 pub env: String,
580 pub gateway: String,
582 pub mode: String,
584 pub node: String,
586 pub transport: String,
588 pub out: String,
590 pub files_written: usize,
593 pub manifest: String,
595 pub render_sha256: String,
598 pub skipped: Vec<Skipped>,
601 pub warnings: Vec<String>,
603 pub excluded_tag_providers: Vec<String>,
607}
608
609struct Existing {
613 files: Vec<String>,
614 dirs: BTreeSet<String>,
616}
617
618fn repo_entry_exists(repo: &Path, rel: &str) -> Result<bool, CoreError> {
620 match std::fs::symlink_metadata(repo.join(rel)) {
621 Ok(_) => Ok(true),
622 Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(false),
623 Err(e) => Err(CoreError::deploy_tree_invalid(
624 rel,
625 None,
626 None,
627 format!("cannot inspect ({})", e.kind()),
628 )),
629 }
630}
631
632fn scan_existing(repo: &Path, gateway: &str, env: &str) -> Result<Option<Existing>, CoreError> {
636 let gateway_rel = format!("gateways/{gateway}");
637 for rel in ["gateways", gateway_rel.as_str()] {
638 if !repo_entry_exists(repo, rel)? {
639 return Ok(None);
640 }
641 require_repo_dir(repo, rel)?;
642 }
643 let mut existing = Existing {
644 files: Vec::new(),
645 dirs: BTreeSet::new(),
646 };
647 for sub in [
648 "resources".to_string(),
649 "tags".to_string(),
650 format!("modes/{env}"),
651 ] {
652 if sub.starts_with("modes/") {
653 let modes = format!("{gateway_rel}/modes");
654 if !repo_entry_exists(repo, &modes)? {
655 continue;
656 }
657 require_repo_dir(repo, &modes)?;
658 }
659 let rel = format!("{gateway_rel}/{sub}");
660 if !repo_entry_exists(repo, &rel)? {
661 continue;
662 }
663 require_repo_dir(repo, &rel)?;
664 let mut files: BTreeMap<String, PathBuf> = BTreeMap::new();
665 let mut dirs: BTreeSet<String> = BTreeSet::new();
666 walk_capped(
667 &repo.join(&rel),
668 &sub,
669 &mut files,
670 &mut dirs,
671 READ_MAX_ENTRIES,
672 )
673 .map_err(|why| CoreError::deploy_tree_invalid(&rel, None, None, why))?;
674 existing.dirs.insert(sub.clone());
675 existing.dirs.extend(dirs);
676 existing.files.extend(files.into_keys());
677 }
678 Ok(Some(existing))
679}
680
681fn pull_prose(gateway: &str, env: &str, existing: &Existing) -> String {
683 let shown: Vec<&str> = existing.files.iter().take(5).map(String::as_str).collect();
684 let more = if existing.files.len() > shown.len() {
685 ", ..."
686 } else {
687 ""
688 };
689 format!(
690 "deploy pull would replace gateways/{gateway}/resources, tags and modes/{env} with the live tree of nodes.master ({} existing files, e.g. {}{more})",
691 existing.files.len(),
692 shown.join(", ")
693 )
694}
695
696pub fn deploy_pull_preview(repo: &Path, env: &str) -> Result<Option<String>, CoreError> {
701 deploy_pull_preview_in(repo, repo, env)
702}
703
704pub fn deploy_pull_preview_in(
707 repo: &Path,
708 targets: &Path,
709 env: &str,
710) -> Result<Option<String>, CoreError> {
711 let (_, target) = load_target(targets, env)?;
712 Ok(scan_existing(repo, &target.gateway, env)?
713 .filter(|existing| !existing.files.is_empty())
714 .map(|existing| pull_prose(&target.gateway, env, &existing)))
715}
716
717fn transport_label(node: &NodeTarget) -> &'static str {
718 match (&node.ssh, &node.local) {
719 (Some(_), _) => "ssh",
720 (None, Some(_)) => "local",
721 (None, None) => "none",
722 }
723}
724
725pub async fn deploy_pull(
731 repo: &Path,
732 env: &str,
733 reader: &dyn TreeReader,
734 yes: bool,
735) -> Result<DeployPullResult, CoreError> {
736 deploy_pull_in(repo, repo, env, reader, yes).await
737}
738
739pub async fn deploy_pull_in(
742 repo: &Path,
743 targets: &Path,
744 env: &str,
745 reader: &dyn TreeReader,
746 yes: bool,
747) -> Result<DeployPullResult, CoreError> {
748 let (_, target) = load_target(targets, env)?;
749 let gateway = target.gateway.clone();
750 let existing = scan_existing(repo, &gateway, env)?;
751 if let Some(existing) = existing.as_ref().filter(|e| !e.files.is_empty())
752 && !yes
753 {
754 return Err(CoreError::ConfirmationRequired {
755 operation: pull_prose(&gateway, env, existing),
756 });
757 }
758
759 let manifest_rel = format!("gateways/{gateway}/manifest.yaml");
762 let mut left_out: Vec<Skipped> = Vec::new();
763 let (manifest, generated, excluded) = if repo_entry_exists(repo, &manifest_rel)? {
764 let bytes = read_repo_file(repo, &manifest_rel, None)?;
765 let manifest = Manifest::parse(&manifest_rel, &bytes)?;
766 let excluded = exclude::effective(&target, Some(&manifest));
769 let contradictions = exclude::contradictions(&excluded, &manifest, &manifest_rel);
770 if !contradictions.is_empty() {
771 return Err(CoreError::deploy_findings(contradictions));
772 }
773 (manifest, None, excluded)
774 } else {
775 let excluded = exclude::effective(&target, None);
776 let mut live_types = reader.list_types(CORE_COLLECTION).await?;
777 live_types.extend(reader.list_types(&target.mode).await?);
778 let mut names = reader.list_tag_providers(CORE_COLLECTION).await?;
782 names.extend(reader.list_tag_providers(&target.mode).await?);
783 let providers: Vec<String> = names
784 .into_iter()
785 .filter(|p| !exclude::is_excluded(p, &excluded))
786 .collect::<BTreeSet<String>>()
787 .into_iter()
788 .collect();
789 let (yaml, skipped) = generate_manifest(&live_types, &providers);
790 left_out = skipped;
791 let manifest = Manifest::parse(&manifest_rel, yaml.as_bytes())?;
792 (manifest, Some(yaml), excluded)
793 };
794
795 let scope = Scope::from_manifest(&manifest, &target.mode, &excluded);
796 let observed = reader.read(&scope).await?;
797 let outcome = pull(&PullInput {
798 gateway: &gateway,
799 env,
800 target: &target,
801 manifest: &manifest,
802 observed: &observed,
803 })?;
804
805 write_pulled(
806 repo,
807 &gateway,
808 existing.as_ref(),
809 &outcome.source,
810 generated.as_deref(),
811 )?;
812
813 let removed = removed_unreadable_warnings(&gateway, env, existing.as_ref(), &outcome.skipped);
814 left_out.extend(outcome.skipped);
815 left_out.sort_by(|a, b| a.path.as_bytes().cmp(b.path.as_bytes()));
816 let mut warnings = outcome.rendered.warnings.clone();
817 warnings.extend(removed);
818 warnings.extend(observed.local_units.iter().map(|unit| {
819 format!(
820 "{unit} also exists in the `local` collection, which the gateway loads over the mode collection; pull did not read the `local` copy"
821 )
822 }));
823 Ok(DeployPullResult {
824 env: env.to_string(),
825 gateway: gateway.clone(),
826 mode: target.mode.clone(),
827 node: "master".to_string(),
828 transport: transport_label(&target.nodes.master).to_string(),
829 out: format!("gateways/{gateway}"),
830 files_written: outcome.source.files.len(),
831 manifest: if generated.is_some() {
832 "generated"
833 } else {
834 "kept"
835 }
836 .to_string(),
837 render_sha256: outcome.rendered.tree.render_sha256(),
838 skipped: left_out,
839 warnings,
840 excluded_tag_providers: excluded,
841 })
842}
843
844const REMOVED_FILES_SHOWN: usize = 10;
847
848fn removed_unreadable_warnings(
858 gateway: &str,
859 env: &str,
860 existing: Option<&Existing>,
861 skipped: &[Skipped],
862) -> Vec<String> {
863 let Some(existing) = existing else {
864 return Vec::new();
865 };
866 let mode_prefix = format!("modes/{env}/");
867 let mut out = Vec::new();
868 for unit in skipped.iter().filter(|s| s.reason == REASON_PATH_TOO_LONG) {
869 let under = format!("{}/", unit.path);
870 let removed: Vec<String> = existing
871 .files
872 .iter()
873 .filter(|file| {
874 let base = file.strip_prefix(&mode_prefix).unwrap_or(file);
875 RelPath::new(base)
876 .ok()
877 .and_then(|p| collection_path(&p).ok())
878 .is_some_and(|c| c.as_str() == unit.path || c.as_str().starts_with(&under))
879 })
880 .map(|file| format!("gateways/{gateway}/{file}"))
881 .collect();
882 if removed.is_empty() {
883 continue;
884 }
885 let take = removed.len().min(REMOVED_FILES_SHOWN);
886 let mut named = removed[..take].join(", ");
887 if removed.len() > take {
888 named.push_str(&format!(" and {} more", removed.len() - take));
889 }
890 out.push(format!(
891 "{} 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",
892 unit.path,
893 removed.len()
894 ));
895 }
896 out
897}
898
899fn safe_component(name: &str) -> bool {
902 !name.is_empty()
903 && name != "."
904 && name != ".."
905 && !name.contains(['/', '\\', '\0'])
906 && !name.chars().any(char::is_control)
907}
908
909fn validate_pulled(gateway: &str, source: &SourceTree) -> Result<(), CoreError> {
915 let refuse = |path: &str, why: String| {
916 CoreError::deploy_tree_invalid(format!("gateways/{gateway}/{path}"), None, None, why)
917 };
918 let mut folded: BTreeMap<String, &str> = BTreeMap::new();
919 let mut dir_folds: BTreeSet<String> = BTreeSet::new();
920 for path in source.files.keys() {
921 let parts: Vec<&str> = path.components().collect();
922 if parts.iter().any(|c| !safe_component(c)) {
923 return Err(refuse(
924 path.as_str(),
925 "has an unsafe path component; refused before anything was changed".to_string(),
926 ));
927 }
928 let fold = path.as_str().to_lowercase();
929 if let Some(other) = folded.insert(fold.clone(), path.as_str()) {
930 return Err(refuse(
931 path.as_str(),
932 format!(
933 "collides with {other} when case is folded; a case-insensitive filesystem keeps only one. Refused before anything was changed"
934 ),
935 ));
936 }
937 for end in fold.match_indices('/').map(|(i, _)| i) {
938 dir_folds.insert(fold[..end].to_string());
939 }
940 }
941 for (fold, path) in &folded {
942 if dir_folds.contains(fold) {
943 return Err(refuse(
944 path,
945 "is both a file and a directory in the pulled tree; refused before anything was changed"
946 .to_string(),
947 ));
948 }
949 }
950 Ok(())
951}
952
953fn write_pulled(
957 repo: &Path,
958 gateway: &str,
959 existing: Option<&Existing>,
960 source: &SourceTree,
961 manifest: Option<&str>,
962) -> Result<(), CoreError> {
963 validate_pulled(gateway, source)?;
964 write_pulled_fs(repo, gateway, existing, source, manifest)
965}
966
967#[cfg(unix)]
971fn write_pulled_fs(
972 repo: &Path,
973 gateway: &str,
974 existing: Option<&Existing>,
975 source: &SourceTree,
976 manifest: Option<&str>,
977) -> Result<(), CoreError> {
978 use crate::deploy::tree::dirfd::{self, FdError, Kind};
979 use std::io::Write;
980
981 let fail = |what: &str, e: FdError| match e {
982 FdError::Changed => CoreError::deploy_tree_invalid(
983 what,
984 None,
985 None,
986 "changed while it was written (a symlink or a non-directory appeared); refused, nothing outside the gateway directory was touched",
987 ),
988 FdError::Io(io) => CoreError::Internal(format!("cannot write {what}: {io}")),
989 other => CoreError::Internal(format!("cannot write {what}: {other:?}")),
990 };
991
992 let repo_real = std::fs::canonicalize(repo).map_err(|e| {
996 CoreError::Internal(format!("cannot resolve the repo {}: {e}", repo.display()))
997 })?;
998 let root = dirfd::open_root(&repo_real).map_err(|e| fail("the repo", e))?;
999 let gateways = dirfd::ensure_dir(&root, "gateways").map_err(|e| fail("gateways", e))?;
1000 let gw_label = format!("gateways/{gateway}");
1001 let gw = dirfd::ensure_dir(&gateways, gateway).map_err(|e| fail(&gw_label, e))?;
1002
1003 if let Some(existing) = existing {
1004 for file in &existing.files {
1005 let parts: Vec<&str> = file.split('/').collect();
1006 let (name, parents) = parts.split_last().ok_or_else(|| {
1007 CoreError::Internal("an existing file has an empty path".to_string())
1008 })?;
1009 let label = format!("{gw_label}/{file}");
1010 let dir = dirfd::walk(&gw, parents).map_err(|e| fail(&label, e))?;
1011 match dirfd::kind_at(&dir, name).map_err(|e| fail(&label, e))? {
1012 Kind::File => dirfd::unlink_at(&dir, name).map_err(|e| fail(&label, e))?,
1013 Kind::Missing => {}
1014 _ => return Err(fail(&label, FdError::Changed)),
1015 }
1016 }
1017 for dir_path in existing.dirs.iter().rev() {
1019 let parts: Vec<&str> = dir_path.split('/').collect();
1020 let Some((name, parents)) = parts.split_last() else {
1021 continue;
1022 };
1023 let label = format!("{gw_label}/{dir_path}");
1024 let parent = dirfd::walk(&gw, parents).map_err(|e| fail(&label, e))?;
1025 match dirfd::rmdir_at(&parent, name) {
1026 Ok(()) | Err(FdError::Missing) => {}
1027 Err(FdError::NotEmpty) => {
1028 return Err(CoreError::deploy_tree_invalid(
1029 label,
1030 None,
1031 None,
1032 "is not empty after the pull removed its files (something wrote into it); nothing more was changed",
1033 ));
1034 }
1035 Err(e) => return Err(fail(&label, e)),
1036 }
1037 }
1038 }
1039
1040 if let Some(text) = manifest {
1041 let label = format!("{gw_label}/manifest.yaml");
1042 let mut file = dirfd::create_new(&gw, "manifest.yaml").map_err(|e| fail(&label, e))?;
1043 file.write_all(text.as_bytes())
1044 .map_err(|e| CoreError::Internal(format!("cannot write {label}: {e}")))?;
1045 }
1046 for (path, bytes) in &source.files {
1047 let label = format!("{gw_label}/{path}");
1048 let parts: Vec<&str> = path.components().collect();
1049 if parts.iter().any(|c| !safe_component(c)) {
1050 return Err(CoreError::deploy_tree_invalid(
1051 label,
1052 None,
1053 None,
1054 "has an unsafe path component; refused",
1055 ));
1056 }
1057 let (name, parents) = parts
1058 .split_last()
1059 .ok_or_else(|| CoreError::Internal("a pulled file has an empty path".to_string()))?;
1060 let dir = dirfd::ensure_walk(&gw, parents).map_err(|e| fail(&label, e))?;
1061 let mut file = dirfd::create_new(&dir, name).map_err(|e| fail(&label, e))?;
1062 file.write_all(bytes)
1063 .map_err(|e| CoreError::Internal(format!("cannot write {label}: {e}")))?;
1064 }
1065 Ok(())
1066}
1067
1068#[cfg(not(unix))]
1072fn write_pulled_fs(
1073 repo: &Path,
1074 gateway: &str,
1075 existing: Option<&Existing>,
1076 source: &SourceTree,
1077 manifest: Option<&str>,
1078) -> Result<(), CoreError> {
1079 use std::io::Write;
1080
1081 let io = |what: &Path, e: std::io::Error| {
1082 CoreError::Internal(format!("cannot write {}: {e}", what.display()))
1083 };
1084 let gw = repo.join("gateways").join(gateway);
1085 std::fs::create_dir_all(&gw).map_err(|e| io(&gw, e))?;
1086 if let Some(existing) = existing {
1087 for file in &existing.files {
1088 let path = file.split('/').fold(gw.clone(), |acc, c| acc.join(c));
1089 std::fs::remove_file(&path).map_err(|e| io(&path, e))?;
1090 }
1091 for dir_path in existing.dirs.iter().rev() {
1092 let path = dir_path.split('/').fold(gw.clone(), |acc, c| acc.join(c));
1093 std::fs::remove_dir(&path).map_err(|e| io(&path, e))?;
1094 }
1095 }
1096 let create = |path: &Path, bytes: &[u8]| -> Result<(), CoreError> {
1097 if let Some(parent) = path.parent() {
1098 std::fs::create_dir_all(parent).map_err(|e| io(parent, e))?;
1099 }
1100 let mut file = std::fs::OpenOptions::new()
1101 .write(true)
1102 .create_new(true)
1103 .open(path)
1104 .map_err(|e| io(path, e))?;
1105 file.write_all(bytes).map_err(|e| io(path, e))
1106 };
1107 if let Some(text) = manifest {
1108 create(&gw.join("manifest.yaml"), text.as_bytes())?;
1109 }
1110 for (path, bytes) in &source.files {
1111 let parts: Vec<&str> = path.components().collect();
1112 if parts.iter().any(|c| !safe_component(c)) {
1113 return Err(CoreError::deploy_tree_invalid(
1114 format!("gateways/{gateway}/{path}"),
1115 None,
1116 None,
1117 "has an unsafe path component; refused",
1118 ));
1119 }
1120 let dest = parts.iter().fold(gw.clone(), |acc, c| acc.join(c));
1121 create(&dest, bytes)?;
1122 }
1123 Ok(())
1124}
1125
1126#[derive(Debug, Clone)]
1128pub struct VerifyOptions {
1129 pub probe_deadline: Duration,
1131 pub probe_interval: Duration,
1133 pub log_settle: Duration,
1136}
1137
1138impl Default for VerifyOptions {
1139 fn default() -> Self {
1140 Self {
1141 probe_deadline: verify::PROBE_DEADLINE,
1142 probe_interval: verify::PROBE_INTERVAL,
1143 log_settle: Duration::from_millis(verify::LOG_SETTLE_MS.unsigned_abs()),
1144 }
1145 }
1146}
1147
1148#[derive(Debug, Clone, PartialEq, Serialize)]
1150pub struct DeployVerifyResult {
1151 pub env: String,
1153 pub gateway: String,
1155 pub mode: String,
1157 pub node: String,
1160 pub project: String,
1162 pub ok: bool,
1164 pub slug: Option<&'static str>,
1166 pub checks: Vec<verify::Check>,
1168 pub resources_checked: usize,
1170 pub warnings: Vec<String>,
1172 pub excluded_tag_providers: Vec<String>,
1175 #[serde(skip_serializing_if = "Option::is_none")]
1178 pub backup: Option<BackupReport>,
1179 #[serde(skip_serializing_if = "Option::is_none")]
1182 pub secrets: Option<SecretNamesReport>,
1183}
1184
1185impl DeployVerifyResult {
1186 pub fn failure_exit_code(&self) -> Option<u8> {
1191 (!self.ok).then_some(6)
1192 }
1193}
1194
1195const LOG_PAGES_MAX: usize = 5;
1197
1198pub async fn deploy_verify(
1231 declared: &Declared,
1232 project: &str,
1233 deploy: &dyn DeployApi,
1234 gateway: &dyn GatewayApi,
1235 reader: &dyn TreeReader,
1236 opts: &VerifyOptions,
1237) -> Result<DeployVerifyResult, CoreError> {
1238 deploy_verify_judged(
1239 declared,
1240 project,
1241 deploy,
1242 gateway,
1243 reader,
1244 opts,
1245 VerifyContext::Standalone,
1246 )
1247 .await
1248}
1249
1250#[derive(Debug, Clone, Copy)]
1253pub enum VerifyContext<'a> {
1254 Standalone,
1256 PostApply(&'a verify::ReadBackBaseline),
1259}
1260
1261pub(crate) async fn deploy_verify_judged(
1263 declared: &Declared,
1264 project: &str,
1265 deploy: &dyn DeployApi,
1266 gateway: &dyn GatewayApi,
1267 reader: &dyn TreeReader,
1268 opts: &VerifyOptions,
1269 ctx: VerifyContext<'_>,
1270) -> Result<DeployVerifyResult, CoreError> {
1271 deploy_verify_planned(declared, project, deploy, gateway, reader, opts, ctx)
1272 .await
1273 .map(|(result, _)| result)
1274}
1275
1276async fn deploy_verify_planned(
1281 declared: &Declared,
1282 project: &str,
1283 deploy: &dyn DeployApi,
1284 gateway: &dyn GatewayApi,
1285 reader: &dyn TreeReader,
1286 opts: &VerifyOptions,
1287 ctx: VerifyContext<'_>,
1288) -> Result<(DeployVerifyResult, BTreeMap<String, PlanStatus>), CoreError> {
1289 let expected =
1290 verify::Expected::from_render(&declared.target.mode, &declared.outcome, &declared.manifest);
1291
1292 let scan = deploy.scan_status(ScanKind::Config).await?;
1293 let scan_read_at = std::time::Instant::now();
1294 let mode = normalize_deployment_mode(deploy.deployment_mode().await?);
1295 let read_back = gather_read_back(deploy, &expected).await?;
1296
1297 let excluded = declared.excluded_tag_providers();
1300 let scope = Scope::from_manifest(&declared.manifest, &declared.target.mode, &excluded);
1301 let live = reader.read(&scope).await?;
1302 let redundancy = Redundancy::of(&declared.target);
1305 let planned = plan_with(
1306 &declared.outcome.tree,
1307 &live,
1308 None,
1309 &scope,
1310 &excluded,
1311 redundancy,
1312 );
1313 let (mode_copies, residue) = if redundancy == Redundancy::Pair {
1319 let dirs = pair_mode_tag_dirs_of(reader, &declared.target.mode, &planned.entries).await?;
1320 let mut mode_copies: BTreeMap<String, Vec<String>> = BTreeMap::new();
1321 for entry in planned
1322 .entries
1323 .iter()
1324 .filter(|entry| expected.tag_units.contains(&entry.key))
1325 {
1326 let paths = pair_mode_tag_files(std::slice::from_ref(entry));
1327 if !paths.is_empty() {
1328 mode_copies.insert(entry.key.clone(), paths);
1329 }
1330 }
1331 let claimed: BTreeSet<&String> = mode_copies.values().flatten().collect();
1332 let others: Vec<String> = pair_mode_tag_files(&planned.entries)
1333 .into_iter()
1334 .filter(|path| !claimed.contains(path))
1335 .collect();
1336 let residue = pair_mode_tags_warning(others, &dirs, &declared.target.mode);
1337 (mode_copies, residue)
1338 } else {
1339 (BTreeMap::new(), None)
1340 };
1341 let files = planned
1342 .entries
1343 .iter()
1344 .filter(|entry| expected.tag_units.contains(&entry.key))
1345 .map(|entry| (entry.key.clone(), entry.status))
1346 .collect();
1347 let file_status = resource_statuses(&expected, &planned.entries);
1348
1349 let logs = if scan.scan_active {
1350 Ok(Vec::new())
1352 } else {
1353 settle_logs(gateway, &scan, scan_read_at, opts.log_settle).await?;
1354 read_logs(gateway, verify::log_window_start(&scan)).await?
1355 };
1356 let probes = if expected.probes.is_empty() {
1357 None
1358 } else {
1359 Some(poll_probes(gateway, project, &expected.probes, opts).await?)
1360 };
1361
1362 let observed = verify::ObservedState {
1363 scan,
1364 mode,
1365 read_back,
1366 files,
1367 logs,
1368 probes,
1369 unreadable: planned
1370 .unreadable
1371 .iter()
1372 .map(|unit| (unit.key.clone(), unit.declared))
1373 .collect(),
1374 mode_copies,
1375 };
1376 let judge = match ctx {
1377 VerifyContext::Standalone => verify::ReadBackJudge::Standalone {
1378 file_status: &file_status,
1379 },
1380 VerifyContext::PostApply(baseline) => verify::ReadBackJudge::Baseline(baseline),
1381 };
1382 let report = verify::evaluate_with(&expected, &observed, &judge);
1383 let mut warnings = report.warnings;
1384 warnings.extend(residue);
1385 let result = DeployVerifyResult {
1386 env: declared.env.clone(),
1387 gateway: declared.target.gateway.clone(),
1388 mode: declared.target.mode.clone(),
1389 node: "master".to_string(),
1390 project: project.to_string(),
1391 ok: report.ok,
1392 slug: report.slug,
1393 checks: report.checks,
1394 resources_checked: report.resources_checked,
1395 warnings,
1396 excluded_tag_providers: excluded,
1397 backup: None,
1398 secrets: None,
1399 };
1400 Ok((result, file_status))
1401}
1402
1403fn resource_statuses(
1406 expected: &verify::Expected,
1407 entries: &[PlanEntry],
1408) -> BTreeMap<String, PlanStatus> {
1409 let keys: BTreeSet<&str> = expected.resources.iter().map(|r| r.key.as_str()).collect();
1410 entries
1411 .iter()
1412 .filter(|entry| keys.contains(entry.key.as_str()))
1413 .map(|entry| (entry.key.clone(), entry.status))
1414 .collect()
1415}
1416
1417#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1419pub enum VerifyNode {
1420 Master,
1423 Backup,
1425 Both,
1427}
1428
1429impl VerifyNode {
1430 pub fn as_str(self) -> &'static str {
1432 match self {
1433 VerifyNode::Master => "master",
1434 VerifyNode::Backup => "backup",
1435 VerifyNode::Both => "both",
1436 }
1437 }
1438}
1439
1440pub type VerifyMasterSurfaces<'a> = (&'a dyn DeployApi, &'a dyn GatewayApi);
1442
1443pub async fn deploy_verify_node(
1471 declared: &Declared,
1472 project: &str,
1473 node: VerifyNode,
1474 master: Option<VerifyMasterSurfaces<'_>>,
1475 master_reader: &dyn TreeReader,
1476 backup: Option<&BackupNode<'_>>,
1477 opts: &VerifyOptions,
1478) -> Result<DeployVerifyResult, CoreError> {
1479 let missing = |what: &str| CoreError::InvalidInput {
1480 reason: format!("deploy verify --node {} needs the {what}", node.as_str()),
1481 };
1482 let (master_result, master_status) = if node == VerifyNode::Backup {
1483 (None, None)
1484 } else {
1485 let (deploy, gateway) = master.ok_or_else(|| missing("master surfaces"))?;
1486 let (result, status) = deploy_verify_planned(
1487 declared,
1488 project,
1489 deploy,
1490 gateway,
1491 master_reader,
1492 opts,
1493 VerifyContext::Standalone,
1494 )
1495 .await?;
1496 (Some(result), Some(status))
1497 };
1498 if node == VerifyNode::Master {
1499 return master_result.ok_or_else(|| missing("master surfaces"));
1500 }
1501 let backup = backup.ok_or_else(|| missing("backup node"))?;
1502 let expected =
1503 verify::Expected::from_render(&declared.target.mode, &declared.outcome, &declared.manifest);
1504 let mut backup_warnings = Vec::new();
1505 let (tick, compared) = verify_backup_tick(
1506 declared,
1507 &expected,
1508 master_reader,
1509 backup,
1510 master_status.as_ref(),
1511 &mut backup_warnings,
1512 )
1513 .await;
1514 let report = verify_report(&tick);
1515
1516 let mut result = match master_result {
1517 Some(mut result) => {
1521 result.warnings.extend(backup_warnings);
1522 result
1523 }
1524 None => DeployVerifyResult {
1525 env: declared.env.clone(),
1526 gateway: declared.target.gateway.clone(),
1527 mode: declared.target.mode.clone(),
1528 node: String::new(),
1529 project: project.to_string(),
1530 ok: true,
1531 slug: None,
1532 checks: Vec::new(),
1533 resources_checked: compared,
1534 warnings: backup_warnings,
1535 excluded_tag_providers: declared.excluded_tag_providers(),
1536 backup: None,
1537 secrets: None,
1538 },
1539 };
1540 let secrets = verify_secret_names(
1543 declared,
1544 master
1545 .map(|(deploy, _)| deploy)
1546 .filter(|_| node != VerifyNode::Backup),
1547 Some(backup),
1548 )
1549 .await?;
1550 let secrets_ok = secrets.nodes.iter().all(|n| n.status != CheckStatus::Fail);
1551 result.node = node.as_str().to_string();
1552 result.ok = result.ok && report.converged && secrets_ok;
1553 result.slug = (!result.ok).then_some(verify::VERIFY_FAILED_SLUG);
1554 result.backup = Some(report);
1555 result.secrets = Some(secrets);
1556 Ok(result)
1557}
1558
1559async fn verify_secret_names(
1566 declared: &Declared,
1567 master: Option<&dyn DeployApi>,
1568 backup: Option<&BackupNode<'_>>,
1569) -> Result<SecretNamesReport, CoreError> {
1570 let referenced = collect_referenced_secrets(&declared.outcome.tree);
1571 let declared_providers = declared_secret_providers(&declared.outcome.tree);
1572 let providers: BTreeSet<&str> = referenced.iter().map(|(p, _)| p.as_str()).collect();
1573 let mut nodes = Vec::with_capacity(2);
1574 if let Some(master) = master {
1575 let mut listings = BTreeMap::new();
1576 for provider in &providers {
1577 listings.insert(
1578 (*provider).to_string(),
1579 master.secret_names(provider).await?,
1580 );
1581 }
1582 nodes.push(check_secret_names(
1583 "master",
1584 &referenced,
1585 &listings,
1586 &declared_providers,
1587 SecretCheckMode::Verify,
1588 ));
1589 }
1590 if let Some(backup) = backup {
1591 let mut listings = BTreeMap::new();
1592 for provider in &providers {
1593 listings.insert(
1594 (*provider).to_string(),
1595 backup.secret_names(provider).await?,
1596 );
1597 }
1598 nodes.push(check_secret_names(
1599 "backup",
1600 &referenced,
1601 &listings,
1602 &declared_providers,
1603 SecretCheckMode::Verify,
1604 ));
1605 }
1606 Ok(SecretNamesReport::ordered(nodes))
1607}
1608
1609async fn verify_backup_tick(
1622 declared: &Declared,
1623 expected: &verify::Expected,
1624 master_reader: &dyn TreeReader,
1625 backup: &BackupNode<'_>,
1626 master_status: Option<&BTreeMap<String, PlanStatus>>,
1627 warnings: &mut Vec<String>,
1628) -> (TickOutcome, usize) {
1629 let signals = match backup.status().await {
1630 Ok(status) => Ok(BackupSignals::from_status(None, &status)),
1631 Err(error) => Err(error.code().to_string()),
1632 };
1633 let mode = match backup.deployment_mode().await {
1634 Ok(mode) => mode_backup_check(expected, mode.as_deref()),
1635 Err(error) => unreachable_check(BackupCheckId::Mode, error.code()),
1636 };
1637 let mut files_warnings = Vec::new();
1638 let (files, files_checked, files_status) = verify_backup_files(
1639 declared,
1640 expected,
1641 master_reader,
1642 backup,
1643 master_status.is_none(),
1644 &mut files_warnings,
1645 )
1646 .await;
1647 let file_status = master_status.or(files_status.as_ref());
1648 let judge = match file_status {
1649 Some(file_status) => verify::ReadBackJudge::Standalone { file_status },
1650 None => verify::ReadBackJudge::Strict,
1651 };
1652 let mut compared = 0;
1653 let read_back = match backup.read_back(expected).await {
1654 Ok(read_back) => {
1655 let (check, count) =
1656 read_back_backup_check_with(expected, &read_back, &judge, warnings);
1657 compared = count;
1658 check
1659 }
1660 Err(error) => unreachable_check(BackupCheckId::ReadBack, error.code()),
1661 };
1662 warnings.extend(files_warnings);
1663 (
1664 TickOutcome {
1665 signals,
1666 exhausted: false,
1667 content: Some(vec![mode, read_back, files]),
1668 files_checked,
1669 stable_ticks: 0,
1670 },
1671 compared,
1672 )
1673}
1674
1675async fn verify_backup_files(
1684 declared: &Declared,
1685 expected: &verify::Expected,
1686 master_reader: &dyn TreeReader,
1687 backup: &BackupNode<'_>,
1688 want_status: bool,
1689 warnings: &mut Vec<String>,
1690) -> (BackupCheck, usize, Option<BTreeMap<String, PlanStatus>>) {
1691 let scope = Scope::from_manifest(
1692 &declared.manifest,
1693 &declared.target.mode,
1694 &declared.excluded_tag_providers(),
1695 );
1696 let backup_files = match backup.read_files(&scope).await {
1697 None => return (files_skipped_check(), 0, None),
1699 Some(Ok(observed)) => observed,
1700 Some(Err(error)) => {
1701 return (
1702 unreachable_check(BackupCheckId::Files, &format!("backup {}", error.code())),
1703 0,
1704 None,
1705 );
1706 }
1707 };
1708 let backup_unreadable = unreadable_units(&backup_files);
1709 let backup_files = effective_files(backup_files);
1710 let master_files = match master_reader.read(&scope).await {
1711 Ok(observed) => observed,
1712 Err(error) => {
1713 return (
1714 unreachable_check(BackupCheckId::Files, &format!("master {}", error.code())),
1715 0,
1716 None,
1717 );
1718 }
1719 };
1720 let status = want_status.then(|| {
1721 let excluded = declared.excluded_tag_providers();
1722 let planned = plan_with(
1723 &declared.outcome.tree,
1724 &master_files,
1725 None,
1726 &scope,
1727 &excluded,
1728 Redundancy::of(&declared.target),
1729 );
1730 resource_statuses(expected, &planned.entries)
1731 });
1732 let master_unreadable = unreadable_units(&master_files);
1733 let master_files = effective_files(master_files);
1734 let present: BTreeSet<_> = declared.outcome.tree.files.keys().cloned().collect();
1735 let absent = BTreeSet::new();
1736 let mut comparison = compare_files(&master_files, &backup_files, &present, &absent);
1737 exempt_unreadable(
1738 &mut comparison,
1739 &present,
1740 &absent,
1741 &[
1742 ("master", &master_unreadable),
1743 ("backup", &backup_unreadable),
1744 ],
1745 );
1746 warnings.extend(unreadable_warnings(&comparison));
1747 (files_check(&comparison), comparison.checked, status)
1748}
1749
1750pub(crate) async fn gather_read_back(
1755 deploy: &dyn DeployApi,
1756 expected: &verify::Expected,
1757) -> Result<BTreeMap<String, verify::Remote>, CoreError> {
1758 let classify = |record: crate::client::config_resources::ResourceRecord| {
1759 if record.collection.as_deref() == Some(verify::SYSTEM_COLLECTION) {
1760 verify::Remote::System
1761 } else {
1762 verify::Remote::Config(record.config)
1763 }
1764 };
1765 let mut lists: BTreeMap<(String, String), Option<Vec<_>>> = BTreeMap::new();
1766 let mut out = BTreeMap::new();
1767 for res in &expected.resources {
1768 let remote = match &res.name {
1769 None => match deploy.resource_singleton(&res.module, &res.ty, None).await {
1770 Ok(record) => classify(record),
1771 Err(CoreError::NotFound { .. }) => verify::Remote::Missing,
1772 Err(e) => return Err(e),
1773 },
1774 Some(name) => {
1775 let type_key = (res.module.clone(), res.ty.clone());
1776 if !lists.contains_key(&type_key) {
1777 let items = match deploy
1778 .resource_list(&res.module, &res.ty, &ListQuery::default())
1779 .await
1780 {
1781 Ok(envelope) => Some(envelope.items),
1782 Err(CoreError::NotFound { .. }) => None,
1783 Err(e) => return Err(e),
1784 };
1785 lists.insert(type_key.clone(), items);
1786 }
1787 lists
1788 .get(&type_key)
1789 .and_then(Option::as_ref)
1790 .and_then(|items| items.iter().find(|r| r.name.as_deref() == Some(name)))
1791 .map_or(verify::Remote::Missing, |r| classify(r.clone()))
1792 }
1793 };
1794 out.insert(res.key.clone(), remote);
1795 }
1796 Ok(out)
1797}
1798
1799async fn settle_logs(
1803 gateway: &dyn GatewayApi,
1804 scan: &ScanStatus,
1805 scan_read_at: std::time::Instant,
1806 settle: Duration,
1807) -> Result<(), CoreError> {
1808 if settle.is_zero() {
1809 return Ok(());
1810 }
1811 let settle_ms = i64::try_from(settle.as_millis()).unwrap_or(i64::MAX);
1812 let probe = LogQuery {
1813 start_time: Some(scan.last_scan_timestamp.saturating_add(settle_ms)),
1814 limit: 1,
1815 ..LogQuery::default()
1816 };
1817 match gateway.logs(&probe).await {
1818 Ok(envelope) if !envelope.items.is_empty() => return Ok(()),
1819 Err(e @ CoreError::Network { .. }) => return Err(e),
1820 Ok(_) | Err(_) => {}
1822 }
1823 tokio::time::sleep(settle.saturating_sub(scan_read_at.elapsed())).await;
1824 Ok(())
1825}
1826
1827async fn read_logs(
1831 gateway: &dyn GatewayApi,
1832 start_time: i64,
1833) -> Result<Result<Vec<verify::LogLine>, String>, CoreError> {
1834 let mut lines = Vec::new();
1835 let mut offset = 0i64;
1836 for _ in 0..LOG_PAGES_MAX {
1837 let query = LogQuery {
1838 start_time: Some(start_time),
1839 min_level: Some("WARN".to_string()),
1840 limit: DEFAULT_LOG_LIMIT,
1841 offset,
1842 ..LogQuery::default()
1843 };
1844 match gateway.logs(&query).await {
1845 Ok(envelope) => {
1846 let count = i64::try_from(envelope.items.len()).unwrap_or(i64::MAX);
1847 lines.extend(envelope.items.into_iter().map(|e| verify::LogLine {
1848 timestamp: e.timestamp,
1849 logger: e.logger_name,
1850 message: e.message,
1851 }));
1852 if count < DEFAULT_LOG_LIMIT {
1853 return Ok(Ok(lines));
1854 }
1855 offset += count;
1856 }
1857 Err(e @ CoreError::Network { .. }) => return Err(e),
1858 Err(e) => return Ok(Err(e.code().to_string())),
1859 }
1860 }
1861 Ok(Err("logs_truncated".to_string()))
1862}
1863
1864async fn poll_probes(
1869 gateway: &dyn GatewayApi,
1870 project: &str,
1871 paths: &[String],
1872 opts: &VerifyOptions,
1873) -> Result<Result<Vec<verify::ProbeReading>, String>, CoreError> {
1874 let interval = opts.probe_interval.max(Duration::from_millis(1));
1875 let started = tokio::time::Instant::now();
1876 loop {
1877 let readings: Vec<verify::ProbeReading> =
1878 match crate::actions::tags::tags_read(gateway, project, paths).await {
1879 Ok(result) => result
1880 .results
1881 .into_iter()
1882 .enumerate()
1883 .map(|(i, row)| verify::ProbeReading {
1884 path: if row.path.is_empty() {
1886 paths.get(i).cloned().unwrap_or_default()
1887 } else {
1888 row.path
1889 },
1890 quality: row.quality,
1891 })
1892 .collect(),
1893 Err(e @ CoreError::Network { .. }) => return Err(e),
1894 Err(e) => return Ok(Err(e.code().to_string())),
1895 };
1896 let all_good = paths.iter().all(|path| {
1897 readings
1898 .iter()
1899 .any(|r| &r.path == path && r.quality.starts_with("Good"))
1900 });
1901 if all_good || started.elapsed() + interval > opts.probe_deadline {
1902 return Ok(Ok(readings));
1903 }
1904 tokio::time::sleep(interval).await;
1905 }
1906}
1907
1908#[cfg(test)]
1909mod tests {
1910 use super::*;
1911
1912 fn repo_with_existing() -> (tempfile::TempDir, Existing) {
1914 let repo = tempfile::TempDir::new().expect("repo");
1915 let dir = repo.path().join("gateways/g/resources/ignition/x");
1916 std::fs::create_dir_all(&dir).expect("dirs");
1917 std::fs::write(dir.join("keep.json"), b"{\"keep\": true}").expect("file");
1918 let existing = Existing {
1919 files: vec!["resources/ignition/x/keep.json".to_string()],
1920 dirs: ["resources", "resources/ignition", "resources/ignition/x"]
1921 .iter()
1922 .map(|d| d.to_string())
1923 .collect(),
1924 };
1925 (repo, existing)
1926 }
1927
1928 fn snapshot(root: &Path) -> BTreeMap<PathBuf, Vec<u8>> {
1929 fn walk(dir: &Path, out: &mut BTreeMap<PathBuf, Vec<u8>>) {
1930 for entry in std::fs::read_dir(dir).expect("read_dir") {
1931 let path = entry.expect("entry").path();
1932 if path.is_dir() {
1933 out.insert(path.clone(), Vec::new());
1934 walk(&path, out);
1935 } else {
1936 out.insert(path.clone(), std::fs::read(&path).expect("read"));
1937 }
1938 }
1939 }
1940 let mut out = BTreeMap::new();
1941 walk(root, &mut out);
1942 out
1943 }
1944
1945 fn assert_refused_untouched(files: &[(&str, &str)], needle: &str) {
1948 let (repo, existing) = repo_with_existing();
1949 let before = snapshot(repo.path());
1950 let source = SourceTree::from_files(files.iter().copied()).expect("source");
1951 let err =
1952 write_pulled(repo.path(), "g", Some(&existing), &source, None).expect_err("refused");
1953 assert_eq!(err.code(), "deploy_tree_invalid", "{err}");
1954 assert!(err.to_string().contains(needle), "{needle}: {err}");
1955 assert_eq!(
1956 snapshot(repo.path()),
1957 before,
1958 "a refused pull must not delete the existing tree"
1959 );
1960 }
1961
1962 #[test]
1963 fn unsafe_component_is_refused_before_any_delete() {
1964 assert_refused_untouched(
1965 &[
1966 ("resources/ignition/x/fine.json", "{}"),
1967 ("resources/ignition/x/bad\nname.json", "{}"),
1968 ],
1969 "unsafe path component",
1970 );
1971 }
1972
1973 #[test]
1974 fn case_fold_collision_is_refused_before_any_delete() {
1975 assert_refused_untouched(
1976 &[
1977 ("resources/ignition/x/Foo.json", "{}"),
1978 ("resources/ignition/x/foo.json", "{}"),
1979 ],
1980 "case is folded",
1981 );
1982 }
1983
1984 #[test]
1985 fn file_and_directory_clash_is_refused_before_any_delete() {
1986 assert_refused_untouched(
1987 &[
1988 ("resources/ignition/x", "{}"),
1989 ("resources/ignition/x/a.json", "{}"),
1990 ],
1991 "both a file and a directory",
1992 );
1993 }
1994
1995 #[test]
1996 fn a_valid_tree_replaces_the_existing_one() {
1997 let (repo, existing) = repo_with_existing();
1998 let source =
1999 SourceTree::from_files([("resources/ignition/x/new.json", "{}")]).expect("source");
2000 write_pulled(repo.path(), "g", Some(&existing), &source, None).expect("written");
2001 let dir = repo.path().join("gateways/g/resources/ignition/x");
2002 assert!(!dir.join("keep.json").exists());
2003 assert!(dir.join("new.json").is_file());
2004 }
2005}