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 #[serde(skip)]
322 pub pair_mode_tag_dirs: Vec<String>,
323}
324
325#[derive(Debug, Clone, Default)]
327pub struct PlanOptions {
328 pub base: Option<PathBuf>,
331 pub diff: bool,
333}
334
335#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
337pub struct BaseSummary {
338 pub dir: String,
340 pub render_sha256: String,
342 pub files: usize,
344}
345
346pub async fn deploy_plan(
350 declared: &Declared,
351 reader: &dyn TreeReader,
352 opts: &PlanOptions,
353) -> Result<DeployPlanResult, CoreError> {
354 let node = &declared.target.nodes.master;
355 let transport = match (&node.ssh, &node.local) {
356 (Some(_), _) => "ssh",
357 (None, Some(_)) => "local",
358 (None, None) => "none",
359 };
360 let base = match &opts.base {
363 Some(dir) => {
364 let dir = dir.clone();
365 let (gateway, env, mode) = (
366 declared.target.gateway.clone(),
367 declared.env.clone(),
368 declared.target.mode.clone(),
369 );
370 let read =
371 tokio::task::spawn_blocking(move || BaseTree::read(&dir, &gateway, &env, &mode))
372 .await
373 .map_err(|e| CoreError::Internal(format!("--base read task failed: {e}")))?;
374 Some(read?)
375 }
376 None => None,
377 };
378 if let Some(journal) = journal_on_box(
383 reader,
384 node,
385 &declared.target.gateway,
386 &declared.target.mode,
387 )
388 .await?
389 {
390 return Err(journal_present(
391 &journal,
392 &declared.env,
393 node,
394 &declared.target.gateway,
395 &declared.target.mode,
396 ));
397 }
398 let excluded = declared.excluded_tag_providers();
399 let scope = Scope::from_manifest(&declared.manifest, &declared.target.mode, &excluded);
400 let observed = reader.read(&scope).await?;
401 let clashes = live_tag_folder_findings(&observed, &declared.target.mode);
405 if !clashes.is_empty() {
406 return Err(CoreError::deploy_findings(clashes));
407 }
408 let gateway = &declared.target.gateway;
409 let mode = &declared.target.mode;
410 let ledger = match reader.read_ledger(gateway, mode).await? {
413 Some(bytes) => Some(Ledger::parse(
414 &ledger_location(LedgerRoot::for_node(node), gateway, mode),
415 &bytes,
416 gateway,
417 mode,
418 )?),
419 None => None,
420 };
421 let mut planned = plan_with(
424 &declared.outcome.tree,
425 &observed,
426 ledger.as_ref(),
427 &scope,
428 &excluded,
429 Redundancy::of(&declared.target),
430 );
431 if let Some(base) = &base {
432 apply_base(&mut planned, base);
433 }
434 if opts.diff {
435 render_diffs(&mut planned, &declared.outcome.tree, &observed);
436 }
437
438 let mut warnings = declared.outcome.warnings.clone();
439 warnings.extend(planned.warnings);
440 let mut pair_mode_tag_dirs = Vec::new();
441 if declared.target.nodes.backup.is_some() {
442 pair_mode_tag_dirs = pair_mode_tag_dirs_of(reader, mode, &planned.entries).await?;
443 warnings.extend(pair_tags_warning(
444 &planned.entries,
445 &planned.delete_set,
446 &pair_mode_tag_dirs,
447 mode,
448 ));
449 }
450 Ok(DeployPlanResult {
451 env: declared.env.clone(),
452 gateway: declared.target.gateway.clone(),
453 mode: declared.target.mode.clone(),
454 node: "master".to_string(),
455 transport: transport.to_string(),
456 render_sha256: declared.outcome.tree.render_sha256(),
457 plan_sha256: planned.plan_sha256,
458 summary: planned.summary,
459 entries: planned.entries,
460 delete_set: planned.delete_set,
461 base: base.map(|b| BaseSummary {
462 dir: b.dir,
463 render_sha256: b.render_sha256,
464 files: b.files.len(),
465 }),
466 warnings,
467 excluded_tag_providers: excluded,
468 unreadable: planned.unreadable,
469 pair_mode_tag_dirs,
470 })
471}
472
473pub(crate) async fn pair_mode_tag_dirs_of(
481 reader: &dyn TreeReader,
482 mode: &str,
483 entries: &[PlanEntry],
484) -> Result<Vec<String>, CoreError> {
485 let dirs: Vec<(String, String)> = PAIR_UNSYNCED_TAG_TYPES
486 .iter()
487 .map(|ty| ("ignition".to_string(), (*ty).to_string()))
488 .collect();
489 let listed = reader.list_tree(mode, &dirs).await?;
490 Ok(pair_mode_tag_dirs(&listed, entries))
491}
492
493fn named_list(items: &[String]) -> String {
495 let shown = items.len().min(crate::error::APPLY_LIST_SHOWN);
496 let mut named = items[..shown].join(", ");
497 if items.len() > shown {
498 named.push_str(&format!(" and {} more", items.len() - shown));
499 }
500 named
501}
502
503fn pair_tags_warning(
515 entries: &[PlanEntry],
516 delete: &DeleteSet,
517 dirs: &[String],
518 mode: &str,
519) -> Option<String> {
520 let mut keys = pair_tag_keys(entries, delete);
521 keys.extend(dirs.iter().cloned());
522 keys.sort_by(|a, b| a.as_bytes().cmp(b.as_bytes()));
523 keys.dedup();
524 if keys.is_empty() {
525 return None;
526 }
527 Some(format!(
528 "redundant pair: {} tag unit(s) or folder(s) would leave files or folders in the {mode} \
529 collection that this deploy cannot delete (the ledger does not list them, the delete \
530 set skips them, or they are files or folders the deploy does not read), which Ignition's \
531 redundancy sync cannot carry; deploy apply refuses this plan \
532 (deploy_pair_tags_refused): {}",
533 keys.len(),
534 named_list(&keys)
535 ))
536}
537
538fn pair_mode_tags_warning(files: Vec<String>, dirs: &[String], mode: &str) -> Option<String> {
551 let mut paths = files;
552 paths.extend(dirs.iter().cloned());
553 if paths.is_empty() {
554 return None;
555 }
556 Some(format!(
557 "redundant pair: the master's {mode} collection holds {} tag file(s) or folder(s), which \
558 Ignition's redundancy sync cannot carry (the backup stops syncing configuration); a \
559 deploy that declares the units and lists the files in its ledger moves them to the core \
560 collection, or remove them on the master (README runbook T6e): {}",
561 paths.len(),
562 named_list(&paths)
563 ))
564}
565
566#[derive(Debug, Clone, PartialEq, Serialize)]
568pub struct DeployPullResult {
569 pub env: String,
571 pub gateway: String,
573 pub mode: String,
575 pub node: String,
577 pub transport: String,
579 pub out: String,
581 pub files_written: usize,
584 pub manifest: String,
586 pub render_sha256: String,
589 pub skipped: Vec<Skipped>,
592 pub warnings: Vec<String>,
594 pub excluded_tag_providers: Vec<String>,
598}
599
600struct Existing {
604 files: Vec<String>,
605 dirs: BTreeSet<String>,
607}
608
609fn repo_entry_exists(repo: &Path, rel: &str) -> Result<bool, CoreError> {
611 match std::fs::symlink_metadata(repo.join(rel)) {
612 Ok(_) => Ok(true),
613 Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(false),
614 Err(e) => Err(CoreError::deploy_tree_invalid(
615 rel,
616 None,
617 None,
618 format!("cannot inspect ({})", e.kind()),
619 )),
620 }
621}
622
623fn scan_existing(repo: &Path, gateway: &str, env: &str) -> Result<Option<Existing>, CoreError> {
627 let gateway_rel = format!("gateways/{gateway}");
628 for rel in ["gateways", gateway_rel.as_str()] {
629 if !repo_entry_exists(repo, rel)? {
630 return Ok(None);
631 }
632 require_repo_dir(repo, rel)?;
633 }
634 let mut existing = Existing {
635 files: Vec::new(),
636 dirs: BTreeSet::new(),
637 };
638 for sub in [
639 "resources".to_string(),
640 "tags".to_string(),
641 format!("modes/{env}"),
642 ] {
643 if sub.starts_with("modes/") {
644 let modes = format!("{gateway_rel}/modes");
645 if !repo_entry_exists(repo, &modes)? {
646 continue;
647 }
648 require_repo_dir(repo, &modes)?;
649 }
650 let rel = format!("{gateway_rel}/{sub}");
651 if !repo_entry_exists(repo, &rel)? {
652 continue;
653 }
654 require_repo_dir(repo, &rel)?;
655 let mut files: BTreeMap<String, PathBuf> = BTreeMap::new();
656 let mut dirs: BTreeSet<String> = BTreeSet::new();
657 walk_capped(
658 &repo.join(&rel),
659 &sub,
660 &mut files,
661 &mut dirs,
662 READ_MAX_ENTRIES,
663 )
664 .map_err(|why| CoreError::deploy_tree_invalid(&rel, None, None, why))?;
665 existing.dirs.insert(sub.clone());
666 existing.dirs.extend(dirs);
667 existing.files.extend(files.into_keys());
668 }
669 Ok(Some(existing))
670}
671
672fn pull_prose(gateway: &str, env: &str, existing: &Existing) -> String {
674 let shown: Vec<&str> = existing.files.iter().take(5).map(String::as_str).collect();
675 let more = if existing.files.len() > shown.len() {
676 ", ..."
677 } else {
678 ""
679 };
680 format!(
681 "deploy pull would replace gateways/{gateway}/resources, tags and modes/{env} with the live tree of nodes.master ({} existing files, e.g. {}{more})",
682 existing.files.len(),
683 shown.join(", ")
684 )
685}
686
687pub fn deploy_pull_preview(repo: &Path, env: &str) -> Result<Option<String>, CoreError> {
692 deploy_pull_preview_in(repo, repo, env)
693}
694
695pub fn deploy_pull_preview_in(
698 repo: &Path,
699 targets: &Path,
700 env: &str,
701) -> Result<Option<String>, CoreError> {
702 let (_, target) = load_target(targets, env)?;
703 Ok(scan_existing(repo, &target.gateway, env)?
704 .filter(|existing| !existing.files.is_empty())
705 .map(|existing| pull_prose(&target.gateway, env, &existing)))
706}
707
708fn transport_label(node: &NodeTarget) -> &'static str {
709 match (&node.ssh, &node.local) {
710 (Some(_), _) => "ssh",
711 (None, Some(_)) => "local",
712 (None, None) => "none",
713 }
714}
715
716pub async fn deploy_pull(
722 repo: &Path,
723 env: &str,
724 reader: &dyn TreeReader,
725 yes: bool,
726) -> Result<DeployPullResult, CoreError> {
727 deploy_pull_in(repo, repo, env, reader, yes).await
728}
729
730pub async fn deploy_pull_in(
733 repo: &Path,
734 targets: &Path,
735 env: &str,
736 reader: &dyn TreeReader,
737 yes: bool,
738) -> Result<DeployPullResult, CoreError> {
739 let (_, target) = load_target(targets, env)?;
740 let gateway = target.gateway.clone();
741 let existing = scan_existing(repo, &gateway, env)?;
742 if let Some(existing) = existing.as_ref().filter(|e| !e.files.is_empty())
743 && !yes
744 {
745 return Err(CoreError::ConfirmationRequired {
746 operation: pull_prose(&gateway, env, existing),
747 });
748 }
749
750 let manifest_rel = format!("gateways/{gateway}/manifest.yaml");
753 let mut left_out: Vec<Skipped> = Vec::new();
754 let (manifest, generated, excluded) = if repo_entry_exists(repo, &manifest_rel)? {
755 let bytes = read_repo_file(repo, &manifest_rel, None)?;
756 let manifest = Manifest::parse(&manifest_rel, &bytes)?;
757 let excluded = exclude::effective(&target, Some(&manifest));
760 let contradictions = exclude::contradictions(&excluded, &manifest, &manifest_rel);
761 if !contradictions.is_empty() {
762 return Err(CoreError::deploy_findings(contradictions));
763 }
764 (manifest, None, excluded)
765 } else {
766 let excluded = exclude::effective(&target, None);
767 let mut live_types = reader.list_types(CORE_COLLECTION).await?;
768 live_types.extend(reader.list_types(&target.mode).await?);
769 let mut names = reader.list_tag_providers(CORE_COLLECTION).await?;
773 names.extend(reader.list_tag_providers(&target.mode).await?);
774 let providers: Vec<String> = names
775 .into_iter()
776 .filter(|p| !exclude::is_excluded(p, &excluded))
777 .collect::<BTreeSet<String>>()
778 .into_iter()
779 .collect();
780 let (yaml, skipped) = generate_manifest(&live_types, &providers);
781 left_out = skipped;
782 let manifest = Manifest::parse(&manifest_rel, yaml.as_bytes())?;
783 (manifest, Some(yaml), excluded)
784 };
785
786 let scope = Scope::from_manifest(&manifest, &target.mode, &excluded);
787 let observed = reader.read(&scope).await?;
788 let outcome = pull(&PullInput {
789 gateway: &gateway,
790 env,
791 target: &target,
792 manifest: &manifest,
793 observed: &observed,
794 })?;
795
796 write_pulled(
797 repo,
798 &gateway,
799 existing.as_ref(),
800 &outcome.source,
801 generated.as_deref(),
802 )?;
803
804 let removed = removed_unreadable_warnings(&gateway, env, existing.as_ref(), &outcome.skipped);
805 left_out.extend(outcome.skipped);
806 left_out.sort_by(|a, b| a.path.as_bytes().cmp(b.path.as_bytes()));
807 let mut warnings = outcome.rendered.warnings.clone();
808 warnings.extend(removed);
809 warnings.extend(observed.local_units.iter().map(|unit| {
810 format!(
811 "{unit} also exists in the `local` collection, which the gateway loads over the mode collection; pull did not read the `local` copy"
812 )
813 }));
814 Ok(DeployPullResult {
815 env: env.to_string(),
816 gateway: gateway.clone(),
817 mode: target.mode.clone(),
818 node: "master".to_string(),
819 transport: transport_label(&target.nodes.master).to_string(),
820 out: format!("gateways/{gateway}"),
821 files_written: outcome.source.files.len(),
822 manifest: if generated.is_some() {
823 "generated"
824 } else {
825 "kept"
826 }
827 .to_string(),
828 render_sha256: outcome.rendered.tree.render_sha256(),
829 skipped: left_out,
830 warnings,
831 excluded_tag_providers: excluded,
832 })
833}
834
835const REMOVED_FILES_SHOWN: usize = 10;
838
839fn removed_unreadable_warnings(
849 gateway: &str,
850 env: &str,
851 existing: Option<&Existing>,
852 skipped: &[Skipped],
853) -> Vec<String> {
854 let Some(existing) = existing else {
855 return Vec::new();
856 };
857 let mode_prefix = format!("modes/{env}/");
858 let mut out = Vec::new();
859 for unit in skipped.iter().filter(|s| s.reason == REASON_PATH_TOO_LONG) {
860 let under = format!("{}/", unit.path);
861 let removed: Vec<String> = existing
862 .files
863 .iter()
864 .filter(|file| {
865 let base = file.strip_prefix(&mode_prefix).unwrap_or(file);
866 RelPath::new(base)
867 .ok()
868 .and_then(|p| collection_path(&p).ok())
869 .is_some_and(|c| c.as_str() == unit.path || c.as_str().starts_with(&under))
870 })
871 .map(|file| format!("gateways/{gateway}/{file}"))
872 .collect();
873 if removed.is_empty() {
874 continue;
875 }
876 let take = removed.len().min(REMOVED_FILES_SHOWN);
877 let mut named = removed[..take].join(", ");
878 if removed.len() > take {
879 named.push_str(&format!(" and {} more", removed.len() - take));
880 }
881 out.push(format!(
882 "{} 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",
883 unit.path,
884 removed.len()
885 ));
886 }
887 out
888}
889
890fn safe_component(name: &str) -> bool {
893 !name.is_empty()
894 && name != "."
895 && name != ".."
896 && !name.contains(['/', '\\', '\0'])
897 && !name.chars().any(char::is_control)
898}
899
900fn validate_pulled(gateway: &str, source: &SourceTree) -> Result<(), CoreError> {
906 let refuse = |path: &str, why: String| {
907 CoreError::deploy_tree_invalid(format!("gateways/{gateway}/{path}"), None, None, why)
908 };
909 let mut folded: BTreeMap<String, &str> = BTreeMap::new();
910 let mut dir_folds: BTreeSet<String> = BTreeSet::new();
911 for path in source.files.keys() {
912 let parts: Vec<&str> = path.components().collect();
913 if parts.iter().any(|c| !safe_component(c)) {
914 return Err(refuse(
915 path.as_str(),
916 "has an unsafe path component; refused before anything was changed".to_string(),
917 ));
918 }
919 let fold = path.as_str().to_lowercase();
920 if let Some(other) = folded.insert(fold.clone(), path.as_str()) {
921 return Err(refuse(
922 path.as_str(),
923 format!(
924 "collides with {other} when case is folded; a case-insensitive filesystem keeps only one. Refused before anything was changed"
925 ),
926 ));
927 }
928 for end in fold.match_indices('/').map(|(i, _)| i) {
929 dir_folds.insert(fold[..end].to_string());
930 }
931 }
932 for (fold, path) in &folded {
933 if dir_folds.contains(fold) {
934 return Err(refuse(
935 path,
936 "is both a file and a directory in the pulled tree; refused before anything was changed"
937 .to_string(),
938 ));
939 }
940 }
941 Ok(())
942}
943
944fn write_pulled(
948 repo: &Path,
949 gateway: &str,
950 existing: Option<&Existing>,
951 source: &SourceTree,
952 manifest: Option<&str>,
953) -> Result<(), CoreError> {
954 validate_pulled(gateway, source)?;
955 write_pulled_fs(repo, gateway, existing, source, manifest)
956}
957
958#[cfg(unix)]
962fn write_pulled_fs(
963 repo: &Path,
964 gateway: &str,
965 existing: Option<&Existing>,
966 source: &SourceTree,
967 manifest: Option<&str>,
968) -> Result<(), CoreError> {
969 use crate::deploy::tree::dirfd::{self, FdError, Kind};
970 use std::io::Write;
971
972 let fail = |what: &str, e: FdError| match e {
973 FdError::Changed => CoreError::deploy_tree_invalid(
974 what,
975 None,
976 None,
977 "changed while it was written (a symlink or a non-directory appeared); refused, nothing outside the gateway directory was touched",
978 ),
979 FdError::Io(io) => CoreError::Internal(format!("cannot write {what}: {io}")),
980 other => CoreError::Internal(format!("cannot write {what}: {other:?}")),
981 };
982
983 let repo_real = std::fs::canonicalize(repo).map_err(|e| {
987 CoreError::Internal(format!("cannot resolve the repo {}: {e}", repo.display()))
988 })?;
989 let root = dirfd::open_root(&repo_real).map_err(|e| fail("the repo", e))?;
990 let gateways = dirfd::ensure_dir(&root, "gateways").map_err(|e| fail("gateways", e))?;
991 let gw_label = format!("gateways/{gateway}");
992 let gw = dirfd::ensure_dir(&gateways, gateway).map_err(|e| fail(&gw_label, e))?;
993
994 if let Some(existing) = existing {
995 for file in &existing.files {
996 let parts: Vec<&str> = file.split('/').collect();
997 let (name, parents) = parts.split_last().ok_or_else(|| {
998 CoreError::Internal("an existing file has an empty path".to_string())
999 })?;
1000 let label = format!("{gw_label}/{file}");
1001 let dir = dirfd::walk(&gw, parents).map_err(|e| fail(&label, e))?;
1002 match dirfd::kind_at(&dir, name).map_err(|e| fail(&label, e))? {
1003 Kind::File => dirfd::unlink_at(&dir, name).map_err(|e| fail(&label, e))?,
1004 Kind::Missing => {}
1005 _ => return Err(fail(&label, FdError::Changed)),
1006 }
1007 }
1008 for dir_path in existing.dirs.iter().rev() {
1010 let parts: Vec<&str> = dir_path.split('/').collect();
1011 let Some((name, parents)) = parts.split_last() else {
1012 continue;
1013 };
1014 let label = format!("{gw_label}/{dir_path}");
1015 let parent = dirfd::walk(&gw, parents).map_err(|e| fail(&label, e))?;
1016 match dirfd::rmdir_at(&parent, name) {
1017 Ok(()) | Err(FdError::Missing) => {}
1018 Err(FdError::NotEmpty) => {
1019 return Err(CoreError::deploy_tree_invalid(
1020 label,
1021 None,
1022 None,
1023 "is not empty after the pull removed its files (something wrote into it); nothing more was changed",
1024 ));
1025 }
1026 Err(e) => return Err(fail(&label, e)),
1027 }
1028 }
1029 }
1030
1031 if let Some(text) = manifest {
1032 let label = format!("{gw_label}/manifest.yaml");
1033 let mut file = dirfd::create_new(&gw, "manifest.yaml").map_err(|e| fail(&label, e))?;
1034 file.write_all(text.as_bytes())
1035 .map_err(|e| CoreError::Internal(format!("cannot write {label}: {e}")))?;
1036 }
1037 for (path, bytes) in &source.files {
1038 let label = format!("{gw_label}/{path}");
1039 let parts: Vec<&str> = path.components().collect();
1040 if parts.iter().any(|c| !safe_component(c)) {
1041 return Err(CoreError::deploy_tree_invalid(
1042 label,
1043 None,
1044 None,
1045 "has an unsafe path component; refused",
1046 ));
1047 }
1048 let (name, parents) = parts
1049 .split_last()
1050 .ok_or_else(|| CoreError::Internal("a pulled file has an empty path".to_string()))?;
1051 let dir = dirfd::ensure_walk(&gw, parents).map_err(|e| fail(&label, e))?;
1052 let mut file = dirfd::create_new(&dir, name).map_err(|e| fail(&label, e))?;
1053 file.write_all(bytes)
1054 .map_err(|e| CoreError::Internal(format!("cannot write {label}: {e}")))?;
1055 }
1056 Ok(())
1057}
1058
1059#[cfg(not(unix))]
1063fn write_pulled_fs(
1064 repo: &Path,
1065 gateway: &str,
1066 existing: Option<&Existing>,
1067 source: &SourceTree,
1068 manifest: Option<&str>,
1069) -> Result<(), CoreError> {
1070 use std::io::Write;
1071
1072 let io = |what: &Path, e: std::io::Error| {
1073 CoreError::Internal(format!("cannot write {}: {e}", what.display()))
1074 };
1075 let gw = repo.join("gateways").join(gateway);
1076 std::fs::create_dir_all(&gw).map_err(|e| io(&gw, e))?;
1077 if let Some(existing) = existing {
1078 for file in &existing.files {
1079 let path = file.split('/').fold(gw.clone(), |acc, c| acc.join(c));
1080 std::fs::remove_file(&path).map_err(|e| io(&path, e))?;
1081 }
1082 for dir_path in existing.dirs.iter().rev() {
1083 let path = dir_path.split('/').fold(gw.clone(), |acc, c| acc.join(c));
1084 std::fs::remove_dir(&path).map_err(|e| io(&path, e))?;
1085 }
1086 }
1087 let create = |path: &Path, bytes: &[u8]| -> Result<(), CoreError> {
1088 if let Some(parent) = path.parent() {
1089 std::fs::create_dir_all(parent).map_err(|e| io(parent, e))?;
1090 }
1091 let mut file = std::fs::OpenOptions::new()
1092 .write(true)
1093 .create_new(true)
1094 .open(path)
1095 .map_err(|e| io(path, e))?;
1096 file.write_all(bytes).map_err(|e| io(path, e))
1097 };
1098 if let Some(text) = manifest {
1099 create(&gw.join("manifest.yaml"), text.as_bytes())?;
1100 }
1101 for (path, bytes) in &source.files {
1102 let parts: Vec<&str> = path.components().collect();
1103 if parts.iter().any(|c| !safe_component(c)) {
1104 return Err(CoreError::deploy_tree_invalid(
1105 format!("gateways/{gateway}/{path}"),
1106 None,
1107 None,
1108 "has an unsafe path component; refused",
1109 ));
1110 }
1111 let dest = parts.iter().fold(gw.clone(), |acc, c| acc.join(c));
1112 create(&dest, bytes)?;
1113 }
1114 Ok(())
1115}
1116
1117#[derive(Debug, Clone)]
1119pub struct VerifyOptions {
1120 pub probe_deadline: Duration,
1122 pub probe_interval: Duration,
1124 pub log_settle: Duration,
1127}
1128
1129impl Default for VerifyOptions {
1130 fn default() -> Self {
1131 Self {
1132 probe_deadline: verify::PROBE_DEADLINE,
1133 probe_interval: verify::PROBE_INTERVAL,
1134 log_settle: Duration::from_millis(verify::LOG_SETTLE_MS.unsigned_abs()),
1135 }
1136 }
1137}
1138
1139#[derive(Debug, Clone, PartialEq, Serialize)]
1141pub struct DeployVerifyResult {
1142 pub env: String,
1144 pub gateway: String,
1146 pub mode: String,
1148 pub node: String,
1151 pub project: String,
1153 pub ok: bool,
1155 pub slug: Option<&'static str>,
1157 pub checks: Vec<verify::Check>,
1159 pub resources_checked: usize,
1161 pub warnings: Vec<String>,
1163 pub excluded_tag_providers: Vec<String>,
1166 #[serde(skip_serializing_if = "Option::is_none")]
1169 pub backup: Option<BackupReport>,
1170 #[serde(skip_serializing_if = "Option::is_none")]
1173 pub secrets: Option<SecretNamesReport>,
1174}
1175
1176impl DeployVerifyResult {
1177 pub fn failure_exit_code(&self) -> Option<u8> {
1182 (!self.ok).then_some(6)
1183 }
1184}
1185
1186const LOG_PAGES_MAX: usize = 5;
1188
1189pub async fn deploy_verify(
1222 declared: &Declared,
1223 project: &str,
1224 deploy: &dyn DeployApi,
1225 gateway: &dyn GatewayApi,
1226 reader: &dyn TreeReader,
1227 opts: &VerifyOptions,
1228) -> Result<DeployVerifyResult, CoreError> {
1229 deploy_verify_judged(
1230 declared,
1231 project,
1232 deploy,
1233 gateway,
1234 reader,
1235 opts,
1236 VerifyContext::Standalone,
1237 )
1238 .await
1239}
1240
1241#[derive(Debug, Clone, Copy)]
1244pub enum VerifyContext<'a> {
1245 Standalone,
1247 PostApply(&'a verify::ReadBackBaseline),
1250}
1251
1252pub(crate) async fn deploy_verify_judged(
1254 declared: &Declared,
1255 project: &str,
1256 deploy: &dyn DeployApi,
1257 gateway: &dyn GatewayApi,
1258 reader: &dyn TreeReader,
1259 opts: &VerifyOptions,
1260 ctx: VerifyContext<'_>,
1261) -> Result<DeployVerifyResult, CoreError> {
1262 deploy_verify_planned(declared, project, deploy, gateway, reader, opts, ctx)
1263 .await
1264 .map(|(result, _)| result)
1265}
1266
1267async fn deploy_verify_planned(
1272 declared: &Declared,
1273 project: &str,
1274 deploy: &dyn DeployApi,
1275 gateway: &dyn GatewayApi,
1276 reader: &dyn TreeReader,
1277 opts: &VerifyOptions,
1278 ctx: VerifyContext<'_>,
1279) -> Result<(DeployVerifyResult, BTreeMap<String, PlanStatus>), CoreError> {
1280 let expected =
1281 verify::Expected::from_render(&declared.target.mode, &declared.outcome, &declared.manifest);
1282
1283 let scan = deploy.scan_status(ScanKind::Config).await?;
1284 let scan_read_at = std::time::Instant::now();
1285 let mode = normalize_deployment_mode(deploy.deployment_mode().await?);
1286 let read_back = gather_read_back(deploy, &expected).await?;
1287
1288 let excluded = declared.excluded_tag_providers();
1291 let scope = Scope::from_manifest(&declared.manifest, &declared.target.mode, &excluded);
1292 let live = reader.read(&scope).await?;
1293 let redundancy = Redundancy::of(&declared.target);
1296 let planned = plan_with(
1297 &declared.outcome.tree,
1298 &live,
1299 None,
1300 &scope,
1301 &excluded,
1302 redundancy,
1303 );
1304 let (mode_copies, residue) = if redundancy == Redundancy::Pair {
1310 let dirs = pair_mode_tag_dirs_of(reader, &declared.target.mode, &planned.entries).await?;
1311 let mut mode_copies: BTreeMap<String, Vec<String>> = BTreeMap::new();
1312 for entry in planned
1313 .entries
1314 .iter()
1315 .filter(|entry| expected.tag_units.contains(&entry.key))
1316 {
1317 let paths = pair_mode_tag_files(std::slice::from_ref(entry));
1318 if !paths.is_empty() {
1319 mode_copies.insert(entry.key.clone(), paths);
1320 }
1321 }
1322 let claimed: BTreeSet<&String> = mode_copies.values().flatten().collect();
1323 let others: Vec<String> = pair_mode_tag_files(&planned.entries)
1324 .into_iter()
1325 .filter(|path| !claimed.contains(path))
1326 .collect();
1327 let residue = pair_mode_tags_warning(others, &dirs, &declared.target.mode);
1328 (mode_copies, residue)
1329 } else {
1330 (BTreeMap::new(), None)
1331 };
1332 let files = planned
1333 .entries
1334 .iter()
1335 .filter(|entry| expected.tag_units.contains(&entry.key))
1336 .map(|entry| (entry.key.clone(), entry.status))
1337 .collect();
1338 let file_status = resource_statuses(&expected, &planned.entries);
1339
1340 let logs = if scan.scan_active {
1341 Ok(Vec::new())
1343 } else {
1344 settle_logs(gateway, &scan, scan_read_at, opts.log_settle).await?;
1345 read_logs(gateway, verify::log_window_start(&scan)).await?
1346 };
1347 let probes = if expected.probes.is_empty() {
1348 None
1349 } else {
1350 Some(poll_probes(gateway, project, &expected.probes, opts).await?)
1351 };
1352
1353 let observed = verify::ObservedState {
1354 scan,
1355 mode,
1356 read_back,
1357 files,
1358 logs,
1359 probes,
1360 unreadable: planned
1361 .unreadable
1362 .iter()
1363 .map(|unit| (unit.key.clone(), unit.declared))
1364 .collect(),
1365 mode_copies,
1366 };
1367 let judge = match ctx {
1368 VerifyContext::Standalone => verify::ReadBackJudge::Standalone {
1369 file_status: &file_status,
1370 },
1371 VerifyContext::PostApply(baseline) => verify::ReadBackJudge::Baseline(baseline),
1372 };
1373 let report = verify::evaluate_with(&expected, &observed, &judge);
1374 let mut warnings = report.warnings;
1375 warnings.extend(residue);
1376 let result = DeployVerifyResult {
1377 env: declared.env.clone(),
1378 gateway: declared.target.gateway.clone(),
1379 mode: declared.target.mode.clone(),
1380 node: "master".to_string(),
1381 project: project.to_string(),
1382 ok: report.ok,
1383 slug: report.slug,
1384 checks: report.checks,
1385 resources_checked: report.resources_checked,
1386 warnings,
1387 excluded_tag_providers: excluded,
1388 backup: None,
1389 secrets: None,
1390 };
1391 Ok((result, file_status))
1392}
1393
1394fn resource_statuses(
1397 expected: &verify::Expected,
1398 entries: &[PlanEntry],
1399) -> BTreeMap<String, PlanStatus> {
1400 let keys: BTreeSet<&str> = expected.resources.iter().map(|r| r.key.as_str()).collect();
1401 entries
1402 .iter()
1403 .filter(|entry| keys.contains(entry.key.as_str()))
1404 .map(|entry| (entry.key.clone(), entry.status))
1405 .collect()
1406}
1407
1408#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1410pub enum VerifyNode {
1411 Master,
1414 Backup,
1416 Both,
1418}
1419
1420impl VerifyNode {
1421 pub fn as_str(self) -> &'static str {
1423 match self {
1424 VerifyNode::Master => "master",
1425 VerifyNode::Backup => "backup",
1426 VerifyNode::Both => "both",
1427 }
1428 }
1429}
1430
1431pub type VerifyMasterSurfaces<'a> = (&'a dyn DeployApi, &'a dyn GatewayApi);
1433
1434pub async fn deploy_verify_node(
1462 declared: &Declared,
1463 project: &str,
1464 node: VerifyNode,
1465 master: Option<VerifyMasterSurfaces<'_>>,
1466 master_reader: &dyn TreeReader,
1467 backup: Option<&BackupNode<'_>>,
1468 opts: &VerifyOptions,
1469) -> Result<DeployVerifyResult, CoreError> {
1470 let missing = |what: &str| CoreError::InvalidInput {
1471 reason: format!("deploy verify --node {} needs the {what}", node.as_str()),
1472 };
1473 let (master_result, master_status) = if node == VerifyNode::Backup {
1474 (None, None)
1475 } else {
1476 let (deploy, gateway) = master.ok_or_else(|| missing("master surfaces"))?;
1477 let (result, status) = deploy_verify_planned(
1478 declared,
1479 project,
1480 deploy,
1481 gateway,
1482 master_reader,
1483 opts,
1484 VerifyContext::Standalone,
1485 )
1486 .await?;
1487 (Some(result), Some(status))
1488 };
1489 if node == VerifyNode::Master {
1490 return master_result.ok_or_else(|| missing("master surfaces"));
1491 }
1492 let backup = backup.ok_or_else(|| missing("backup node"))?;
1493 let expected =
1494 verify::Expected::from_render(&declared.target.mode, &declared.outcome, &declared.manifest);
1495 let mut backup_warnings = Vec::new();
1496 let (tick, compared) = verify_backup_tick(
1497 declared,
1498 &expected,
1499 master_reader,
1500 backup,
1501 master_status.as_ref(),
1502 &mut backup_warnings,
1503 )
1504 .await;
1505 let report = verify_report(&tick);
1506
1507 let mut result = match master_result {
1508 Some(mut result) => {
1512 result.warnings.extend(backup_warnings);
1513 result
1514 }
1515 None => DeployVerifyResult {
1516 env: declared.env.clone(),
1517 gateway: declared.target.gateway.clone(),
1518 mode: declared.target.mode.clone(),
1519 node: String::new(),
1520 project: project.to_string(),
1521 ok: true,
1522 slug: None,
1523 checks: Vec::new(),
1524 resources_checked: compared,
1525 warnings: backup_warnings,
1526 excluded_tag_providers: declared.excluded_tag_providers(),
1527 backup: None,
1528 secrets: None,
1529 },
1530 };
1531 let secrets = verify_secret_names(
1534 declared,
1535 master
1536 .map(|(deploy, _)| deploy)
1537 .filter(|_| node != VerifyNode::Backup),
1538 Some(backup),
1539 )
1540 .await?;
1541 let secrets_ok = secrets.nodes.iter().all(|n| n.status != CheckStatus::Fail);
1542 result.node = node.as_str().to_string();
1543 result.ok = result.ok && report.converged && secrets_ok;
1544 result.slug = (!result.ok).then_some(verify::VERIFY_FAILED_SLUG);
1545 result.backup = Some(report);
1546 result.secrets = Some(secrets);
1547 Ok(result)
1548}
1549
1550async fn verify_secret_names(
1557 declared: &Declared,
1558 master: Option<&dyn DeployApi>,
1559 backup: Option<&BackupNode<'_>>,
1560) -> Result<SecretNamesReport, CoreError> {
1561 let referenced = collect_referenced_secrets(&declared.outcome.tree);
1562 let declared_providers = declared_secret_providers(&declared.outcome.tree);
1563 let providers: BTreeSet<&str> = referenced.iter().map(|(p, _)| p.as_str()).collect();
1564 let mut nodes = Vec::with_capacity(2);
1565 if let Some(master) = master {
1566 let mut listings = BTreeMap::new();
1567 for provider in &providers {
1568 listings.insert(
1569 (*provider).to_string(),
1570 master.secret_names(provider).await?,
1571 );
1572 }
1573 nodes.push(check_secret_names(
1574 "master",
1575 &referenced,
1576 &listings,
1577 &declared_providers,
1578 SecretCheckMode::Verify,
1579 ));
1580 }
1581 if let Some(backup) = backup {
1582 let mut listings = BTreeMap::new();
1583 for provider in &providers {
1584 listings.insert(
1585 (*provider).to_string(),
1586 backup.secret_names(provider).await?,
1587 );
1588 }
1589 nodes.push(check_secret_names(
1590 "backup",
1591 &referenced,
1592 &listings,
1593 &declared_providers,
1594 SecretCheckMode::Verify,
1595 ));
1596 }
1597 Ok(SecretNamesReport::ordered(nodes))
1598}
1599
1600async fn verify_backup_tick(
1613 declared: &Declared,
1614 expected: &verify::Expected,
1615 master_reader: &dyn TreeReader,
1616 backup: &BackupNode<'_>,
1617 master_status: Option<&BTreeMap<String, PlanStatus>>,
1618 warnings: &mut Vec<String>,
1619) -> (TickOutcome, usize) {
1620 let signals = match backup.status().await {
1621 Ok(status) => Ok(BackupSignals::from_status(None, &status)),
1622 Err(error) => Err(error.code().to_string()),
1623 };
1624 let mode = match backup.deployment_mode().await {
1625 Ok(mode) => mode_backup_check(expected, mode.as_deref()),
1626 Err(error) => unreachable_check(BackupCheckId::Mode, error.code()),
1627 };
1628 let mut files_warnings = Vec::new();
1629 let (files, files_checked, files_status) = verify_backup_files(
1630 declared,
1631 expected,
1632 master_reader,
1633 backup,
1634 master_status.is_none(),
1635 &mut files_warnings,
1636 )
1637 .await;
1638 let file_status = master_status.or(files_status.as_ref());
1639 let judge = match file_status {
1640 Some(file_status) => verify::ReadBackJudge::Standalone { file_status },
1641 None => verify::ReadBackJudge::Strict,
1642 };
1643 let mut compared = 0;
1644 let read_back = match backup.read_back(expected).await {
1645 Ok(read_back) => {
1646 let (check, count) =
1647 read_back_backup_check_with(expected, &read_back, &judge, warnings);
1648 compared = count;
1649 check
1650 }
1651 Err(error) => unreachable_check(BackupCheckId::ReadBack, error.code()),
1652 };
1653 warnings.extend(files_warnings);
1654 (
1655 TickOutcome {
1656 signals,
1657 exhausted: false,
1658 content: Some(vec![mode, read_back, files]),
1659 files_checked,
1660 stable_ticks: 0,
1661 },
1662 compared,
1663 )
1664}
1665
1666async fn verify_backup_files(
1675 declared: &Declared,
1676 expected: &verify::Expected,
1677 master_reader: &dyn TreeReader,
1678 backup: &BackupNode<'_>,
1679 want_status: bool,
1680 warnings: &mut Vec<String>,
1681) -> (BackupCheck, usize, Option<BTreeMap<String, PlanStatus>>) {
1682 let scope = Scope::from_manifest(
1683 &declared.manifest,
1684 &declared.target.mode,
1685 &declared.excluded_tag_providers(),
1686 );
1687 let backup_files = match backup.read_files(&scope).await {
1688 None => return (files_skipped_check(), 0, None),
1690 Some(Ok(observed)) => observed,
1691 Some(Err(error)) => {
1692 return (
1693 unreachable_check(BackupCheckId::Files, &format!("backup {}", error.code())),
1694 0,
1695 None,
1696 );
1697 }
1698 };
1699 let backup_unreadable = unreadable_units(&backup_files);
1700 let backup_files = effective_files(backup_files);
1701 let master_files = match master_reader.read(&scope).await {
1702 Ok(observed) => observed,
1703 Err(error) => {
1704 return (
1705 unreachable_check(BackupCheckId::Files, &format!("master {}", error.code())),
1706 0,
1707 None,
1708 );
1709 }
1710 };
1711 let status = want_status.then(|| {
1712 let excluded = declared.excluded_tag_providers();
1713 let planned = plan_with(
1714 &declared.outcome.tree,
1715 &master_files,
1716 None,
1717 &scope,
1718 &excluded,
1719 Redundancy::of(&declared.target),
1720 );
1721 resource_statuses(expected, &planned.entries)
1722 });
1723 let master_unreadable = unreadable_units(&master_files);
1724 let master_files = effective_files(master_files);
1725 let present: BTreeSet<_> = declared.outcome.tree.files.keys().cloned().collect();
1726 let absent = BTreeSet::new();
1727 let mut comparison = compare_files(&master_files, &backup_files, &present, &absent);
1728 exempt_unreadable(
1729 &mut comparison,
1730 &present,
1731 &absent,
1732 &[
1733 ("master", &master_unreadable),
1734 ("backup", &backup_unreadable),
1735 ],
1736 );
1737 warnings.extend(unreadable_warnings(&comparison));
1738 (files_check(&comparison), comparison.checked, status)
1739}
1740
1741pub(crate) async fn gather_read_back(
1746 deploy: &dyn DeployApi,
1747 expected: &verify::Expected,
1748) -> Result<BTreeMap<String, verify::Remote>, CoreError> {
1749 let classify = |record: crate::client::config_resources::ResourceRecord| {
1750 if record.collection.as_deref() == Some(verify::SYSTEM_COLLECTION) {
1751 verify::Remote::System
1752 } else {
1753 verify::Remote::Config(record.config)
1754 }
1755 };
1756 let mut lists: BTreeMap<(String, String), Option<Vec<_>>> = BTreeMap::new();
1757 let mut out = BTreeMap::new();
1758 for res in &expected.resources {
1759 let remote = match &res.name {
1760 None => match deploy.resource_singleton(&res.module, &res.ty, None).await {
1761 Ok(record) => classify(record),
1762 Err(CoreError::NotFound { .. }) => verify::Remote::Missing,
1763 Err(e) => return Err(e),
1764 },
1765 Some(name) => {
1766 let type_key = (res.module.clone(), res.ty.clone());
1767 if !lists.contains_key(&type_key) {
1768 let items = match deploy
1769 .resource_list(&res.module, &res.ty, &ListQuery::default())
1770 .await
1771 {
1772 Ok(envelope) => Some(envelope.items),
1773 Err(CoreError::NotFound { .. }) => None,
1774 Err(e) => return Err(e),
1775 };
1776 lists.insert(type_key.clone(), items);
1777 }
1778 lists
1779 .get(&type_key)
1780 .and_then(Option::as_ref)
1781 .and_then(|items| items.iter().find(|r| r.name.as_deref() == Some(name)))
1782 .map_or(verify::Remote::Missing, |r| classify(r.clone()))
1783 }
1784 };
1785 out.insert(res.key.clone(), remote);
1786 }
1787 Ok(out)
1788}
1789
1790async fn settle_logs(
1794 gateway: &dyn GatewayApi,
1795 scan: &ScanStatus,
1796 scan_read_at: std::time::Instant,
1797 settle: Duration,
1798) -> Result<(), CoreError> {
1799 if settle.is_zero() {
1800 return Ok(());
1801 }
1802 let settle_ms = i64::try_from(settle.as_millis()).unwrap_or(i64::MAX);
1803 let probe = LogQuery {
1804 start_time: Some(scan.last_scan_timestamp.saturating_add(settle_ms)),
1805 limit: 1,
1806 ..LogQuery::default()
1807 };
1808 match gateway.logs(&probe).await {
1809 Ok(envelope) if !envelope.items.is_empty() => return Ok(()),
1810 Err(e @ CoreError::Network { .. }) => return Err(e),
1811 Ok(_) | Err(_) => {}
1813 }
1814 tokio::time::sleep(settle.saturating_sub(scan_read_at.elapsed())).await;
1815 Ok(())
1816}
1817
1818async fn read_logs(
1822 gateway: &dyn GatewayApi,
1823 start_time: i64,
1824) -> Result<Result<Vec<verify::LogLine>, String>, CoreError> {
1825 let mut lines = Vec::new();
1826 let mut offset = 0i64;
1827 for _ in 0..LOG_PAGES_MAX {
1828 let query = LogQuery {
1829 start_time: Some(start_time),
1830 min_level: Some("WARN".to_string()),
1831 limit: DEFAULT_LOG_LIMIT,
1832 offset,
1833 ..LogQuery::default()
1834 };
1835 match gateway.logs(&query).await {
1836 Ok(envelope) => {
1837 let count = i64::try_from(envelope.items.len()).unwrap_or(i64::MAX);
1838 lines.extend(envelope.items.into_iter().map(|e| verify::LogLine {
1839 timestamp: e.timestamp,
1840 logger: e.logger_name,
1841 message: e.message,
1842 }));
1843 if count < DEFAULT_LOG_LIMIT {
1844 return Ok(Ok(lines));
1845 }
1846 offset += count;
1847 }
1848 Err(e @ CoreError::Network { .. }) => return Err(e),
1849 Err(e) => return Ok(Err(e.code().to_string())),
1850 }
1851 }
1852 Ok(Err("logs_truncated".to_string()))
1853}
1854
1855async fn poll_probes(
1860 gateway: &dyn GatewayApi,
1861 project: &str,
1862 paths: &[String],
1863 opts: &VerifyOptions,
1864) -> Result<Result<Vec<verify::ProbeReading>, String>, CoreError> {
1865 let interval = opts.probe_interval.max(Duration::from_millis(1));
1866 let started = tokio::time::Instant::now();
1867 loop {
1868 let readings: Vec<verify::ProbeReading> =
1869 match crate::actions::tags::tags_read(gateway, project, paths).await {
1870 Ok(result) => result
1871 .results
1872 .into_iter()
1873 .enumerate()
1874 .map(|(i, row)| verify::ProbeReading {
1875 path: if row.path.is_empty() {
1877 paths.get(i).cloned().unwrap_or_default()
1878 } else {
1879 row.path
1880 },
1881 quality: row.quality,
1882 })
1883 .collect(),
1884 Err(e @ CoreError::Network { .. }) => return Err(e),
1885 Err(e) => return Ok(Err(e.code().to_string())),
1886 };
1887 let all_good = paths.iter().all(|path| {
1888 readings
1889 .iter()
1890 .any(|r| &r.path == path && r.quality.starts_with("Good"))
1891 });
1892 if all_good || started.elapsed() + interval > opts.probe_deadline {
1893 return Ok(Ok(readings));
1894 }
1895 tokio::time::sleep(interval).await;
1896 }
1897}
1898
1899#[cfg(test)]
1900mod tests {
1901 use super::*;
1902
1903 fn repo_with_existing() -> (tempfile::TempDir, Existing) {
1905 let repo = tempfile::TempDir::new().expect("repo");
1906 let dir = repo.path().join("gateways/g/resources/ignition/x");
1907 std::fs::create_dir_all(&dir).expect("dirs");
1908 std::fs::write(dir.join("keep.json"), b"{\"keep\": true}").expect("file");
1909 let existing = Existing {
1910 files: vec!["resources/ignition/x/keep.json".to_string()],
1911 dirs: ["resources", "resources/ignition", "resources/ignition/x"]
1912 .iter()
1913 .map(|d| d.to_string())
1914 .collect(),
1915 };
1916 (repo, existing)
1917 }
1918
1919 fn snapshot(root: &Path) -> BTreeMap<PathBuf, Vec<u8>> {
1920 fn walk(dir: &Path, out: &mut BTreeMap<PathBuf, Vec<u8>>) {
1921 for entry in std::fs::read_dir(dir).expect("read_dir") {
1922 let path = entry.expect("entry").path();
1923 if path.is_dir() {
1924 out.insert(path.clone(), Vec::new());
1925 walk(&path, out);
1926 } else {
1927 out.insert(path.clone(), std::fs::read(&path).expect("read"));
1928 }
1929 }
1930 }
1931 let mut out = BTreeMap::new();
1932 walk(root, &mut out);
1933 out
1934 }
1935
1936 fn assert_refused_untouched(files: &[(&str, &str)], needle: &str) {
1939 let (repo, existing) = repo_with_existing();
1940 let before = snapshot(repo.path());
1941 let source = SourceTree::from_files(files.iter().copied()).expect("source");
1942 let err =
1943 write_pulled(repo.path(), "g", Some(&existing), &source, None).expect_err("refused");
1944 assert_eq!(err.code(), "deploy_tree_invalid", "{err}");
1945 assert!(err.to_string().contains(needle), "{needle}: {err}");
1946 assert_eq!(
1947 snapshot(repo.path()),
1948 before,
1949 "a refused pull must not delete the existing tree"
1950 );
1951 }
1952
1953 #[test]
1954 fn unsafe_component_is_refused_before_any_delete() {
1955 assert_refused_untouched(
1956 &[
1957 ("resources/ignition/x/fine.json", "{}"),
1958 ("resources/ignition/x/bad\nname.json", "{}"),
1959 ],
1960 "unsafe path component",
1961 );
1962 }
1963
1964 #[test]
1965 fn case_fold_collision_is_refused_before_any_delete() {
1966 assert_refused_untouched(
1967 &[
1968 ("resources/ignition/x/Foo.json", "{}"),
1969 ("resources/ignition/x/foo.json", "{}"),
1970 ],
1971 "case is folded",
1972 );
1973 }
1974
1975 #[test]
1976 fn file_and_directory_clash_is_refused_before_any_delete() {
1977 assert_refused_untouched(
1978 &[
1979 ("resources/ignition/x", "{}"),
1980 ("resources/ignition/x/a.json", "{}"),
1981 ],
1982 "both a file and a directory",
1983 );
1984 }
1985
1986 #[test]
1987 fn a_valid_tree_replaces_the_existing_one() {
1988 let (repo, existing) = repo_with_existing();
1989 let source =
1990 SourceTree::from_files([("resources/ignition/x/new.json", "{}")]).expect("source");
1991 write_pulled(repo.path(), "g", Some(&existing), &source, None).expect("written");
1992 let dir = repo.path().join("gateways/g/resources/ignition/x");
1993 assert!(!dir.join("keep.json").exists());
1994 assert!(dir.join("new.json").is_file());
1995 }
1996}