1use std::io::Write as _;
104use std::path::PathBuf;
105use std::time::{Duration, Instant};
106
107use async_trait::async_trait;
108use serde::Serialize;
109
110use super::deploy::load_target;
111use super::projects::validate_import;
112use super::resources::export_zip_bytes;
113use super::workspace::rfc3339_now_utc;
114use crate::client::GatewayApi;
115use crate::client::projects::{ImportOutcome, ProjectRecord};
116use crate::client::query::ListQuery;
117use crate::deploy::apply::hold::SystemClock;
118use crate::deploy::apply::preflight::{
119 DesignerSeen, DesignersPolicy, DesignersVerdict, designers_detail, designers_gate,
120 read_designers, wrong_target_failure, wrong_target_gate,
121};
122use crate::deploy::apply::run_store::{EventLog, RunDir, RunId, run_dir_rel};
123use crate::deploy::exclude;
124use crate::deploy::ledger::{LedgerRoot, release_record_location};
125use crate::deploy::project_artifact::{
126 Artifact, CanonicalDigest, build_artifact, canonical_digest,
127};
128use crate::deploy::project_tags::{
129 FILES_PER_ENTRY, Lookup, MISSING_TAGS_LIMIT, MissingTag, TagRef, TagRefs, TagTree,
130 TagsUnavailable, UnknownProvider, scan_artifact,
131};
132use crate::deploy::projects::{
133 MemberDelta, PROJECT_FILE_REL, ParentLink, ProjectAction, ProjectFile, ProjectMeta, Release,
134 ReleaseOrder, ReleaseRecord, check_parent, load_project_file, member_delta, parent_chain,
135 read_project_meta,
136};
137use crate::deploy::target::{NodeTarget, Target};
138use crate::deploy::transport::for_node;
139use crate::deploy::transport::projects::{
140 ProjectHost, ReleaseRecordWriter, git_entry_location, release_writer_for_node,
141};
142use crate::deploy::transport::remote::safe_remote_path;
143use crate::error::{CoreError, PreflightFailure};
144
145pub const LIST_CAP: usize = 200;
147
148pub const CHILDREN_SETTLE_S: u64 = 30;
152
153pub const CHILDREN_POLL_S: u64 = 5;
155
156pub const PROJECTS_CANARY_EXEMPT: &[&str] = &["export"];
166
167#[async_trait]
174pub trait ProjectGateway: Send + Sync {
175 async fn projects(&self) -> Result<Vec<ProjectRecord>, CoreError>;
177 async fn project_find(&self, name: &str) -> Result<Option<ProjectRecord>, CoreError>;
179 async fn export_bytes(&self, name: &str) -> Result<Vec<u8>, CoreError>;
181 async fn designers(&self) -> Result<Vec<DesignerSeen>, CoreError>;
184 async fn gateway_name(&self) -> Result<Option<String>, CoreError>;
188 async fn tag_provider_names(&self) -> Result<Vec<String>, CoreError>;
191 async fn tag_export(&self, provider: &str) -> Result<Vec<u8>, CoreError>;
194 async fn project_import(
198 &self,
199 name: &str,
200 zip: Vec<u8>,
201 overwrite: bool,
202 ) -> Result<ImportOutcome, CoreError>;
203}
204
205pub struct GatewayProjects<'a>(pub &'a dyn GatewayApi);
207
208#[async_trait]
209impl ProjectGateway for GatewayProjects<'_> {
210 async fn projects(&self) -> Result<Vec<ProjectRecord>, CoreError> {
211 Ok(self.0.projects(&ListQuery::default()).await?.items)
212 }
213
214 async fn project_find(&self, name: &str) -> Result<Option<ProjectRecord>, CoreError> {
215 match self.0.project_find(name).await {
216 Ok(record) => Ok(Some(record)),
217 Err(CoreError::NotFound { .. }) => Ok(None),
218 Err(err) => Err(err),
219 }
220 }
221
222 async fn export_bytes(&self, name: &str) -> Result<Vec<u8>, CoreError> {
223 export_zip_bytes(self.0, name).await
224 }
225
226 async fn designers(&self) -> Result<Vec<DesignerSeen>, CoreError> {
227 read_designers(self.0).await
228 }
229
230 async fn gateway_name(&self) -> Result<Option<String>, CoreError> {
231 Ok(self
232 .0
233 .gateway_info()
234 .await?
235 .name
236 .filter(|name| !name.trim().is_empty()))
237 }
238
239 async fn tag_provider_names(&self) -> Result<Vec<String>, CoreError> {
240 Ok(self
241 .0
242 .tag_provider_list(&ListQuery::default())
243 .await?
244 .items
245 .into_iter()
246 .map(|record| record.name)
247 .collect())
248 }
249
250 async fn tag_export(&self, provider: &str) -> Result<Vec<u8>, CoreError> {
251 self.0.tags_export_provider(provider).await
252 }
253
254 async fn project_import(
255 &self,
256 name: &str,
257 zip: Vec<u8>,
258 overwrite: bool,
259 ) -> Result<ImportOutcome, CoreError> {
260 self.0.project_import(name, zip, overwrite).await
261 }
262}
263
264pub struct ProjectsSurfaces<'a> {
269 pub gateway: &'a dyn ProjectGateway,
271 pub host: &'a dyn ProjectHost,
273}
274
275#[derive(Debug, Clone)]
277pub struct ProjectsInput {
278 pub repo: PathBuf,
280 pub targets: PathBuf,
282 pub env: String,
284 pub release: String,
286 pub allow_downgrade: bool,
289 pub allow_first_change: Option<String>,
292 pub artifact_sha256: Option<String>,
297}
298
299pub const FIRST_CHANGE_REASON_MAX: usize = 500;
301
302#[derive(Debug)]
304pub struct LoadedProject {
305 pub file: ProjectFile,
307 pub meta: ProjectMeta,
309 pub target_file: String,
311 pub target: Target,
313 pub artifact: Artifact,
315 pub release: Release,
317}
318
319pub fn load_project(input: &ProjectsInput) -> Result<LoadedProject, CoreError> {
321 let release = Release::parse(&input.release)?;
324 if let Some(reason) = &input.allow_first_change {
325 let chars = reason.chars().count();
326 if chars == 0 || chars > FIRST_CHANGE_REASON_MAX {
327 return Err(CoreError::InvalidInput {
328 reason: format!(
329 "--allow-first-change <reason>: the reason is 1 to {FIRST_CHANGE_REASON_MAX} characters"
330 ),
331 });
332 }
333 }
334 if let Some(hash) = &input.artifact_sha256
335 && !is_lower_hex64(hash)
336 {
337 return Err(CoreError::InvalidInput {
338 reason: "--artifact-sha256 must be 64 lowercase hex characters (the \
339 `artifact_sha256` of `ign deploy projects --check --json`)"
340 .to_string(),
341 });
342 }
343 let file = load_project_file(&input.repo)?;
344 file.require_env(&input.env)?;
345 let meta = read_project_meta(&input.repo)?;
346 let (target_file, target) = load_target(&input.targets, &input.env)?;
347 record_names_fit_the_host(
350 &file.project,
351 &target.gateway,
352 &target_file,
353 &target.nodes.master,
354 )?;
355 let artifact = build_artifact(&input.repo)?;
356 check_artifact_zip(&artifact.zip)?;
357 if let Some(expected) = &input.artifact_sha256
361 && *expected != artifact.digest.sha256
362 {
363 return Err(CoreError::ProjectHashMismatch {
364 project: file.project.clone(),
365 expected: expected.clone(),
366 actual: artifact.digest.sha256.clone(),
367 });
368 }
369 Ok(LoadedProject {
370 file,
371 meta,
372 target_file,
373 target,
374 artifact,
375 release,
376 })
377}
378
379fn record_names_fit_the_host(
388 project: &str,
389 gateway: &str,
390 target_file: &str,
391 master: &NodeTarget,
392) -> Result<(), CoreError> {
393 if master.ssh.is_none() {
394 return Ok(());
395 }
396 if let Err(why) = safe_remote_path(project) {
397 return Err(CoreError::deploy_tree_invalid(
398 PROJECT_FILE_REL,
399 None,
400 None,
401 format!(
402 "project: the name cannot be used for the release record on an ssh node ({why})"
403 ),
404 ));
405 }
406 if let Err(why) = safe_remote_path(gateway) {
407 return Err(CoreError::deploy_tree_invalid(
408 target_file,
409 None,
410 None,
411 format!(
412 "gateway: the name cannot be used for the release record on an ssh node ({why})"
413 ),
414 ));
415 }
416 Ok(())
417}
418
419fn is_lower_hex64(text: &str) -> bool {
421 text.len() == 64
422 && text
423 .bytes()
424 .all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b))
425}
426
427pub(crate) fn check_artifact_zip(zip: &[u8]) -> Result<(), CoreError> {
432 validate_import(zip)
433}
434
435#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
437#[serde(rename_all = "snake_case")]
438pub enum ProjectOutcome {
439 Checked,
441 Skipped,
444 Created,
446 Imported,
449 Recorded,
452 ImportedUnverified,
455}
456
457impl ProjectOutcome {
458 pub fn as_str(self) -> &'static str {
460 match self {
461 ProjectOutcome::Checked => "checked",
462 ProjectOutcome::Skipped => "skipped",
463 ProjectOutcome::Created => "created",
464 ProjectOutcome::Imported => "imported",
465 ProjectOutcome::Recorded => "recorded",
466 ProjectOutcome::ImportedUnverified => "imported_unverified",
467 }
468 }
469}
470
471pub const PROJECT_UNVERIFIED_SLUG: &str = "deploy_project_unverified";
477
478#[derive(Debug, Clone, Serialize)]
480pub struct DeployProjectsResult {
481 pub env: String,
483 pub gateway: String,
485 pub project: String,
487 pub release: String,
489 pub check: bool,
491 pub action: ProjectAction,
493 pub would_change: bool,
495 pub artifact_sha256: String,
497 pub live_sha256: Option<String>,
499 pub members: MemberDelta,
501 pub removed: Vec<String>,
503 pub removed_count: usize,
505 pub excluded: Vec<String>,
510 pub excluded_count: usize,
512 pub unmanaged: Vec<String>,
515 pub unmanaged_count: usize,
517 pub parent: Option<String>,
520 pub previous_release: Option<String>,
523 pub recorded_sha256: Option<String>,
526 pub live_matches_last_deploy: Option<bool>,
529 pub allow_downgrade: bool,
531 pub allow_first_change: Option<String>,
533 pub warnings: Vec<String>,
535 pub run_id: Option<String>,
537 pub run_dir: Option<String>,
540 pub record_written: bool,
542 pub restore_command: Option<String>,
546 pub children: Vec<ChildReport>,
551 pub missing_tags: Option<Vec<MissingTag>>,
557 pub missing_tags_count: Option<usize>,
559 pub missing_tags_unknown_providers: Option<Vec<UnknownProvider>>,
564 pub missing_tags_unknown_providers_count: Option<usize>,
566 pub missing_tags_unavailable: Option<Vec<TagsUnavailable>>,
568 pub missing_tags_limit: &'static str,
570 pub outcome: ProjectOutcome,
572 pub ok: bool,
574 pub slug: Option<String>,
576}
577
578#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
580#[serde(rename_all = "snake_case")]
581pub enum ChildStatus {
582 Ok,
584 Mismatch,
586 Unverified,
589}
590
591impl ChildStatus {
592 pub fn as_str(self) -> &'static str {
594 match self {
595 ChildStatus::Ok => "ok",
596 ChildStatus::Mismatch => "mismatch",
597 ChildStatus::Unverified => "unverified",
598 }
599 }
600}
601
602#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
605pub struct ChildReport {
606 pub name: String,
608 pub status: ChildStatus,
610 pub release: Option<String>,
612 pub recorded_sha256: Option<String>,
614 pub live_sha256: Option<String>,
617}
618
619impl DeployProjectsResult {
620 pub fn failure_exit_code(&self) -> Option<u8> {
624 (self.outcome == ProjectOutcome::ImportedUnverified).then_some(6)
625 }
626}
627
628fn shell_word(text: &str) -> String {
631 let plain = !text.is_empty()
632 && text.chars().all(|c| {
633 c.is_ascii_alphanumeric() || matches!(c, '_' | '-' | '.' | '/' | ':' | '@' | '+' | ',')
634 });
635 if plain {
636 text.to_string()
637 } else {
638 format!("'{}'", text.replace('\'', "'\\''"))
639 }
640}
641
642fn restore_command(project: &str, run_dir_rel: &str) -> String {
647 format!(
648 "ign project import {} --file {} --collision-policy overwrite --yes",
649 shell_word(project),
650 shell_word(&format!("{run_dir_rel}/export/{project}.zip")),
651 )
652}
653
654fn capped(mut items: Vec<String>) -> (Vec<String>, usize) {
655 let count = items.len();
656 items.truncate(LIST_CAP);
657 (items, count)
658}
659
660fn cap_list<T>(mut items: Vec<T>) -> (Vec<T>, usize) {
662 let count = items.len();
663 items.truncate(LIST_CAP);
664 (items, count)
665}
666
667#[derive(Debug, Clone, Default, PartialEq, Eq)]
670pub struct TagCheck {
671 pub missing: Option<Vec<MissingTag>>,
676 pub unknown_providers: Option<Vec<UnknownProvider>>,
680 pub unavailable: Vec<TagsUnavailable>,
683}
684
685fn unavailable_reason(what: &str, err: &CoreError) -> String {
687 format!("{what} ({})", err.code())
688}
689
690const UNRESOLVED_PATHS_SHOWN: usize = 3;
692
693fn unresolved_reason(paths: &std::collections::BTreeSet<String>) -> String {
698 let shown: Vec<&str> = paths
699 .iter()
700 .take(UNRESOLVED_PATHS_SHOWN)
701 .map(String::as_str)
702 .collect();
703 let rest = paths.len() - shown.len();
704 let mut reason = format!(
705 "the provider's tag export does not resolve the UDT type of {} tag path(s): {}",
706 paths.len(),
707 shown.join(", ")
708 );
709 if rest > 0 {
710 reason.push_str(&format!(" and {rest} more"));
711 }
712 reason
713}
714
715fn strip_property(path: &str) -> Option<&str> {
718 let start = path.rfind('/').map_or(0, |at| at + 1);
719 let dot = path[start..].find('.')?;
720 (dot > 0).then(|| &path[..start + dot])
721}
722
723fn provider_shaped(name: &str) -> bool {
728 name.chars().any(char::is_alphabetic)
729 && name
730 .chars()
731 .all(|c| c.is_alphanumeric() || matches!(c, '_' | '-' | ' '))
732 && !name.starts_with(' ')
733 && !name.ends_with(' ')
734 && !name.eq_ignore_ascii_case("client")
735}
736
737struct UnknownAcc {
739 provider: String,
740 paths: std::collections::BTreeSet<String>,
741 files: std::collections::BTreeSet<String>,
742}
743
744struct MissingAcc {
746 path: String,
747 files: std::collections::BTreeSet<String>,
748}
749
750pub async fn check_tag_references(
763 zip: &[u8],
764 excluded: &[String],
765 gateway: &dyn ProjectGateway,
766) -> TagCheck {
767 let refs = match scan_artifact(zip) {
768 Ok(refs) => refs,
769 Err(err) => {
770 return TagCheck {
771 missing: None,
772 unknown_providers: None,
773 unavailable: vec![TagsUnavailable {
774 provider: None,
775 reason: unavailable_reason("the project files could not be scanned", &err),
776 }],
777 };
778 }
779 };
780 let mut check = check_refs(&refs, excluded, gateway).await;
781 if let Some(reason) = refs.cut_reason() {
782 check.unavailable.insert(
783 0,
784 TagsUnavailable {
785 provider: None,
786 reason,
787 },
788 );
789 check.missing = check.missing.filter(|list| !list.is_empty());
790 check.unknown_providers = check.unknown_providers.filter(|list| !list.is_empty());
791 }
792 check
793}
794
795async fn check_refs(refs: &TagRefs, excluded: &[String], gateway: &dyn ProjectGateway) -> TagCheck {
797 let mut by_provider: std::collections::BTreeMap<String, Vec<&TagRef>> =
801 std::collections::BTreeMap::new();
802 for tag in refs.iter() {
803 let fold = tag.provider.to_ascii_lowercase();
804 if exclude::is_excluded(&fold, excluded) {
805 continue;
806 }
807 by_provider.entry(fold).or_default().push(tag);
808 }
809 if by_provider.is_empty() {
810 return TagCheck {
811 missing: Some(Vec::new()),
812 unknown_providers: Some(Vec::new()),
813 unavailable: Vec::new(),
814 };
815 }
816 let names = match gateway.tag_provider_names().await {
817 Ok(names) => names,
818 Err(err) => {
819 return TagCheck {
820 missing: None,
821 unknown_providers: None,
822 unavailable: vec![TagsUnavailable {
823 provider: None,
824 reason: unavailable_reason(
825 "the gateway's tag providers could not be listed",
826 &err,
827 ),
828 }],
829 };
830 }
831 };
832 let mut listed: std::collections::BTreeMap<String, String> = std::collections::BTreeMap::new();
833 for name in names {
834 let fold = name.to_ascii_lowercase();
835 match listed.get(&fold) {
836 Some(kept) if kept.as_bytes() <= name.as_bytes() => {}
837 _ => {
838 listed.insert(fold, name);
839 }
840 }
841 }
842
843 let mut found: std::collections::BTreeMap<(String, String), MissingAcc> =
844 std::collections::BTreeMap::new();
845 let mut unknown: std::collections::BTreeMap<String, UnknownAcc> =
846 std::collections::BTreeMap::new();
847 let mut unavailable = Vec::new();
848 let (mut needed, mut checked) = (0usize, 0usize);
850 for (fold, tags) in &by_provider {
851 let Some(provider) = listed.get(fold) else {
852 if provider_shaped(fold) {
855 let acc = unknown.entry(fold.clone()).or_insert_with(|| UnknownAcc {
856 provider: tags[0].provider.clone(),
857 paths: std::collections::BTreeSet::new(),
858 files: std::collections::BTreeSet::new(),
859 });
860 for tag in tags {
861 if tag.provider < acc.provider {
862 acc.provider = tag.provider.clone();
863 }
864 let path = strip_property(&tag.path).unwrap_or(&tag.path);
865 acc.paths.insert(path.to_lowercase());
866 acc.files.extend(tag.files.iter().cloned());
867 while acc.files.len() > FILES_PER_ENTRY {
868 acc.files.pop_last();
869 }
870 }
871 }
872 continue;
873 };
874 needed += 1;
875 let bytes = match gateway.tag_export(provider).await {
876 Ok(bytes) => bytes,
877 Err(err) => {
878 unavailable.push(TagsUnavailable {
879 provider: Some(provider.clone()),
880 reason: unavailable_reason("the provider's tag export could not be read", &err),
881 });
882 continue;
883 }
884 };
885 let tree = match TagTree::parse(&bytes, provider) {
886 Ok(tree) => tree,
887 Err(err) => {
888 unavailable.push(TagsUnavailable {
889 provider: Some(provider.clone()),
890 reason: unavailable_reason("the provider's tag export is not a tag tree", &err),
891 });
892 continue;
893 }
894 };
895 drop(bytes);
896 let mut unresolved: std::collections::BTreeSet<String> = std::collections::BTreeSet::new();
899 for tag in tags {
900 let full = tree.lookup(&tag.path);
901 let answer = match (full, strip_property(&tag.path)) {
902 (Lookup::Found, _) => Lookup::Found,
903 (other, None) => other,
904 (full, Some(stripped)) => match (full, tree.lookup(stripped)) {
905 (_, Lookup::Found) => Lookup::Found,
906 (Lookup::Missing, Lookup::Missing) => Lookup::Missing,
907 _ => Lookup::Unknown,
908 },
909 };
910 let reported = match answer {
911 Lookup::Found => {
912 checked += 1;
913 continue;
914 }
915 Lookup::Unknown => {
916 unresolved.insert(format!("[{}]{}", tag.provider, tag.path));
917 continue;
918 }
919 Lookup::Missing => {
920 checked += 1;
921 strip_property(&tag.path).unwrap_or(&tag.path)
922 }
923 };
924 let display = format!("[{}]{}", tag.provider, reported);
925 let acc = found
926 .entry((fold.clone(), reported.to_lowercase()))
927 .or_insert_with(|| MissingAcc {
928 path: display.clone(),
929 files: std::collections::BTreeSet::new(),
930 });
931 if display < acc.path {
932 acc.path = display;
933 }
934 acc.files.extend(tag.files.iter().cloned());
935 while acc.files.len() > FILES_PER_ENTRY {
936 acc.files.pop_last();
937 }
938 }
939 if !unresolved.is_empty() {
940 unavailable.push(TagsUnavailable {
941 provider: Some(provider.clone()),
942 reason: unresolved_reason(&unresolved),
943 });
944 }
945 }
946 let missing = if needed > 0 && checked == 0 {
947 None
948 } else {
949 let mut list: Vec<MissingTag> = found
950 .into_values()
951 .map(|acc| MissingTag {
952 path: acc.path,
953 files: acc.files.into_iter().collect(),
954 })
955 .collect();
956 list.sort_by(|a, b| a.path.as_bytes().cmp(b.path.as_bytes()));
957 Some(list)
958 };
959 let mut unknown_providers: Vec<UnknownProvider> = unknown
960 .into_values()
961 .map(|acc| UnknownProvider {
962 provider: acc.provider,
963 path_count: acc.paths.len(),
964 files: acc.files.into_iter().collect(),
965 })
966 .collect();
967 unknown_providers.sort_by(|a, b| a.provider.as_bytes().cmp(b.provider.as_bytes()));
968 TagCheck {
969 missing,
970 unknown_providers: Some(unknown_providers),
971 unavailable,
972 }
973}
974
975const WARNING_PROVIDERS_SHOWN: usize = 5;
977
978fn tag_warnings(project: &str, gateway: &str, check: &TagCheck) -> Vec<String> {
981 let mut warnings = Vec::new();
982 let missing = check.missing.as_ref().map_or(0, Vec::len);
983 if missing > 0 {
984 warnings.push(format!(
985 "project {project}: {missing} tag path(s) the project uses are not on gateway {gateway}; \
986 ign deploy projects does not deploy tags, and paths built at runtime aren't checked"
987 ));
988 }
989 if let Some(unknown) = check.unknown_providers.as_ref().filter(|u| !u.is_empty()) {
990 let shown: Vec<&str> = unknown
991 .iter()
992 .take(WARNING_PROVIDERS_SHOWN)
993 .map(|u| u.provider.as_str())
994 .collect();
995 let rest = unknown.len().saturating_sub(WARNING_PROVIDERS_SHOWN);
996 warnings.push(format!(
997 "project {project}: tag paths the project uses name {} provider(s) that gateway {gateway} does not have: {}{}; \
998 ign deploy projects does not deploy tags, and paths built at runtime aren't checked",
999 unknown.len(),
1000 shown.join(", "),
1001 if rest > 0 {
1002 format!(" and {rest} more")
1003 } else {
1004 String::new()
1005 }
1006 ));
1007 }
1008 for entry in &check.unavailable {
1009 warnings.push(format!(
1010 "project {project}: tag paths were not checked for {}: {}",
1011 entry.provider.as_deref().unwrap_or("every provider"),
1012 entry.reason
1013 ));
1014 }
1015 warnings
1016}
1017
1018fn parent_link(record: &ProjectRecord) -> ParentLink {
1020 ParentLink {
1021 name: record.name.clone(),
1022 parent: record.parent.clone().filter(|p| !p.is_empty()),
1023 inheritable: record.inheritable,
1024 invalid_parent: record.invalid_parent,
1025 }
1026}
1027
1028fn live_digest(project: &str, zip: &[u8]) -> Result<CanonicalDigest, CoreError> {
1031 canonical_digest(zip).map_err(|err| match err {
1032 CoreError::DeployTreeInvalid(_) => CoreError::Internal(format!(
1033 "the gateway's export of project {project} holds no members"
1034 )),
1035 other => other,
1036 })
1037}
1038
1039#[derive(Debug, Clone)]
1041pub enum ProjectsPrepared {
1042 Ready {
1045 prose: String,
1048 plan: DeployProjectsResult,
1051 },
1052 Unchanged(DeployProjectsResult),
1055}
1056
1057impl ProjectsPrepared {
1058 pub fn into_check_result(self) -> DeployProjectsResult {
1061 let (ProjectsPrepared::Ready {
1062 plan: mut result, ..
1063 }
1064 | ProjectsPrepared::Unchanged(mut result)) = self;
1065 result.check = true;
1066 result.outcome = ProjectOutcome::Checked;
1067 result
1068 }
1069}
1070
1071fn plan_prose(plan: &DeployProjectsResult) -> String {
1076 let action = match plan.action {
1077 ProjectAction::Create => "create project",
1078 ProjectAction::Import => "overwrite-import project",
1079 ProjectAction::Record => "record the release only for project",
1080 ProjectAction::Skip => "leave project",
1081 };
1082 let mut lines = vec![format!(
1083 "deploy projects will {action} {} on gateway {} (env {}, release {})",
1084 plan.project, plan.gateway, plan.env, plan.release
1085 )];
1086 if plan.action == ProjectAction::Record {
1087 lines.push("the project itself is not changed".to_string());
1088 } else {
1089 lines.push(format!(
1090 "members: {} added, {} changed, {} removed, {} same",
1091 plan.members.added, plan.members.changed, plan.members.removed, plan.members.same
1092 ));
1093 }
1094 if plan.removed_count > 0 {
1095 lines.push(format!(
1096 "the import deletes {} member(s) the repo does not have:",
1097 plan.removed_count
1098 ));
1099 for path in plan.removed.iter().take(PROSE_REMOVED_SHOWN) {
1100 lines.push(format!(" {path}"));
1101 }
1102 if plan.removed_count > PROSE_REMOVED_SHOWN {
1103 lines.push(format!(
1104 " ... and {} more",
1105 plan.removed_count - PROSE_REMOVED_SHOWN
1106 ));
1107 }
1108 }
1109 let missing = plan.missing_tags_count.unwrap_or(0);
1112 if missing > 0 {
1113 lines.push(format!(
1114 "{missing} tag path(s) the project uses are not on gateway {} (this verb does not deploy tags):",
1115 plan.gateway
1116 ));
1117 for tag in plan.missing_tags.iter().flatten().take(PROSE_REMOVED_SHOWN) {
1118 lines.push(format!(" {}", tag.path));
1119 }
1120 if missing > PROSE_REMOVED_SHOWN {
1121 lines.push(format!(" ... and {} more", missing - PROSE_REMOVED_SHOWN));
1122 }
1123 }
1124 let unknown = plan.missing_tags_unknown_providers_count.unwrap_or(0);
1125 if unknown > 0 {
1126 lines.push(format!(
1127 "{unknown} provider(s) the project uses are not on gateway {}:",
1128 plan.gateway
1129 ));
1130 for entry in plan
1131 .missing_tags_unknown_providers
1132 .iter()
1133 .flatten()
1134 .take(PROSE_REMOVED_SHOWN)
1135 {
1136 lines.push(format!(" {}", entry.provider));
1137 }
1138 if unknown > PROSE_REMOVED_SHOWN {
1139 lines.push(format!(" ... and {} more", unknown - PROSE_REMOVED_SHOWN));
1140 }
1141 }
1142 for entry in plan.missing_tags_unavailable.iter().flatten() {
1143 lines.push(format!(
1144 "tags NOT CHECKED for {}: {}",
1145 entry.provider.as_deref().unwrap_or("every provider"),
1146 entry.reason
1147 ));
1148 }
1149 if plan.action == ProjectAction::Import {
1150 lines.push("the live project is saved into a run directory before the import".to_string());
1151 }
1152 lines.push(format!("artifact_sha256 {}", plan.artifact_sha256));
1153 lines.join("\n")
1154}
1155
1156const PROSE_REMOVED_SHOWN: usize = 20;
1158
1159pub async fn deploy_projects_prepare(
1168 loaded: &LoadedProject,
1169 input: &ProjectsInput,
1170 surfaces: &ProjectsSurfaces<'_>,
1171) -> Result<ProjectsPrepared, CoreError> {
1172 let project = &loaded.file.project;
1173 let gateway = surfaces.gateway;
1174 let gateway_name = &loaded.target.gateway;
1175
1176 let git = surfaces.host.git_entry(project).await?;
1177 if git.is_present() {
1178 return Err(CoreError::ProjectGitManaged {
1179 project: project.clone(),
1180 location: git_entry_location(project),
1181 kind: git.as_str().to_string(),
1182 });
1183 }
1184
1185 let record = match surfaces
1189 .host
1190 .read_release_record(gateway_name, project)
1191 .await?
1192 {
1193 None => None,
1194 Some(bytes) => {
1195 let location = release_record_location(
1196 LedgerRoot::for_node(&loaded.target.nodes.master),
1197 gateway_name,
1198 project,
1199 );
1200 Some(ReleaseRecord::parse(
1201 &bytes,
1202 &location,
1203 gateway_name,
1204 project,
1205 )?)
1206 }
1207 };
1208 if let Some(record) = &record
1209 && loaded.release.order_against(&record.release()?) == ReleaseOrder::Older
1210 && !input.allow_downgrade
1211 {
1212 return Err(CoreError::DestructiveChangeRefused {
1213 gate: "downgrade".to_string(),
1214 keys: vec![project.clone()],
1215 count: 1,
1216 });
1217 }
1218
1219 let records = gateway.projects().await?;
1220 let mut warnings = Vec::new();
1224 if let Some(parent) = &loaded.meta.parent {
1225 let all: Vec<ParentLink> = records.iter().map(parent_link).collect();
1226 let (chain, bound_hit) = parent_chain(parent, &all);
1227 check_parent(project, parent, &chain, bound_hit).map_err(|reason| {
1228 CoreError::ProjectParentInvalid {
1229 project: project.clone(),
1230 parent: parent.clone(),
1231 reason: reason.as_str().to_string(),
1232 }
1233 })?;
1234 if chain
1238 .first()
1239 .is_some_and(|p| p.invalid_parent == Some(true))
1240 {
1241 warnings.push(format!(
1242 "project {project}: the gateway reports that parent {parent} has an invalid parent of its own"
1243 ));
1244 }
1245 }
1246 let repo_parent = loaded.meta.parent.as_deref().filter(|p| !p.is_empty());
1250 if let Some(live_record) = records.iter().find(|r| r.name == *project) {
1251 let live_parent = live_record.parent.as_deref().filter(|p| !p.is_empty());
1252 if live_parent != repo_parent {
1253 warnings.push(format!(
1254 "project {project}: importing this project changes its parent from {} to {}",
1255 live_parent.unwrap_or("none"),
1256 repo_parent.unwrap_or("none")
1257 ));
1258 }
1259 }
1260 let mut others: Vec<String> = records
1261 .into_iter()
1262 .map(|r| r.name)
1263 .filter(|name| name != project)
1264 .collect();
1265 others.sort_by(|a, b| a.as_bytes().cmp(b.as_bytes()));
1266 others.dedup();
1267
1268 let live = match gateway.project_find(project).await? {
1269 None => None,
1270 Some(_) => {
1271 let zip = gateway.export_bytes(project).await?;
1272 Some(live_digest(project, &zip)?)
1273 }
1274 };
1275
1276 let artifact = &loaded.artifact.digest;
1277 let equal = live.as_ref().is_some_and(|live| {
1278 !live.members.is_empty() && !artifact.members.is_empty() && live.sha256 == artifact.sha256
1279 });
1280 let action = match &live {
1281 None => ProjectAction::Create,
1282 Some(_) if equal => match &record {
1285 Some(record) if record.release()?.version == loaded.release.version => {
1286 ProjectAction::Skip
1287 }
1288 _ => ProjectAction::Record,
1289 },
1290 Some(_) => ProjectAction::Import,
1291 };
1292 if action == ProjectAction::Import && record.is_none() && input.allow_first_change.is_none() {
1295 return Err(CoreError::DestructiveChangeRefused {
1296 gate: "first_deploy".to_string(),
1297 keys: vec![project.clone()],
1298 count: 1,
1299 });
1300 }
1301 let tag_check = check_tag_references(
1305 &loaded.artifact.zip,
1306 &exclude::effective(&loaded.target, None),
1307 gateway,
1308 )
1309 .await;
1310 let live_matches_last_deploy = match (&record, &live) {
1311 (Some(record), Some(live)) => Some(record.artifact_sha256 == live.sha256),
1312 _ => None,
1313 };
1314 if let (Some(record), Some(false)) = (&record, live_matches_last_deploy) {
1315 warnings.push(format!(
1316 "project {project}: the live project no longer matches the last release this tool deployed ({})",
1317 record.release
1318 ));
1319 }
1320 warnings.extend(tag_warnings(project, gateway_name, &tag_check));
1321 let (members, removed) = member_delta(artifact, live.as_ref());
1322 let (removed, removed_count) = capped(removed);
1323 let (excluded, excluded_count) = capped(loaded.artifact.excluded.clone());
1324 let (unmanaged, unmanaged_count) = capped(others);
1325 let (missing_tags, missing_tags_count) = match tag_check.missing {
1326 Some(list) => {
1327 let (list, count) = cap_list(list);
1328 (Some(list), Some(count))
1329 }
1330 None => (None, None),
1331 };
1332 let (missing_tags_unknown_providers, missing_tags_unknown_providers_count) =
1333 match tag_check.unknown_providers {
1334 Some(list) => {
1335 let (list, count) = cap_list(list);
1336 (Some(list), Some(count))
1337 }
1338 None => (None, None),
1339 };
1340 let missing_tags_unavailable =
1341 (!tag_check.unavailable.is_empty()).then_some(tag_check.unavailable);
1342
1343 let plan = DeployProjectsResult {
1344 env: input.env.clone(),
1345 gateway: loaded.target.gateway.clone(),
1346 project: project.clone(),
1347 release: input.release.clone(),
1348 check: false,
1349 action,
1350 would_change: action.would_change(),
1351 artifact_sha256: artifact.sha256.clone(),
1352 live_sha256: live.map(|l| l.sha256),
1353 members,
1354 removed,
1355 removed_count,
1356 excluded,
1357 excluded_count,
1358 unmanaged,
1359 unmanaged_count,
1360 parent: loaded.meta.parent.clone(),
1361 previous_release: record.as_ref().map(|r| r.release.clone()),
1362 recorded_sha256: record.map(|r| r.artifact_sha256),
1363 live_matches_last_deploy,
1364 allow_downgrade: input.allow_downgrade,
1365 allow_first_change: input.allow_first_change.clone(),
1366 warnings,
1367 run_id: None,
1368 run_dir: None,
1369 record_written: false,
1370 restore_command: None,
1371 children: Vec::new(),
1372 missing_tags,
1373 missing_tags_count,
1374 missing_tags_unknown_providers,
1375 missing_tags_unknown_providers_count,
1376 missing_tags_unavailable,
1377 missing_tags_limit: MISSING_TAGS_LIMIT,
1378 outcome: match action {
1379 ProjectAction::Create => ProjectOutcome::Created,
1380 ProjectAction::Skip => ProjectOutcome::Skipped,
1381 ProjectAction::Import => ProjectOutcome::Imported,
1382 ProjectAction::Record => ProjectOutcome::Recorded,
1383 },
1384 ok: true,
1385 slug: None,
1386 };
1387 Ok(if action == ProjectAction::Skip {
1388 ProjectsPrepared::Unchanged(plan)
1389 } else {
1390 ProjectsPrepared::Ready {
1391 prose: plan_prose(&plan),
1392 plan,
1393 }
1394 })
1395}
1396
1397#[derive(Debug, Clone, PartialEq, Eq)]
1402pub enum ProjectsEvent {
1403 DesignersPoll {
1406 open: usize,
1408 },
1409 DesignersProceed {
1411 sessions: Vec<DesignerSeen>,
1413 },
1414 DesignersSkipped,
1417 RunStarted {
1419 run_id: String,
1421 action: ProjectAction,
1423 },
1424 PreImageExported {
1426 bytes: usize,
1428 },
1429 ImportSent {
1431 overwrite: bool,
1433 },
1434 ImportAnswered,
1436 Verified {
1438 equal: bool,
1440 },
1441 RecordWritten,
1443 RecordFailed {
1445 code: String,
1447 },
1448 ChildrenSettling {
1451 children: Vec<String>,
1453 },
1454 ChildChecked {
1456 name: String,
1458 status: ChildStatus,
1460 },
1461 Finished {
1463 outcome: ProjectOutcome,
1465 },
1466}
1467
1468impl ProjectsEvent {
1469 pub fn name(&self) -> &'static str {
1471 match self {
1472 ProjectsEvent::DesignersPoll { .. } => "designers_poll",
1473 ProjectsEvent::DesignersProceed { .. } => "designers_proceed",
1474 ProjectsEvent::DesignersSkipped => "designers_skipped",
1475 ProjectsEvent::RunStarted { .. } => "run_started",
1476 ProjectsEvent::PreImageExported { .. } => "pre_image_exported",
1477 ProjectsEvent::ImportSent { .. } => "import_sent",
1478 ProjectsEvent::ImportAnswered => "import_answered",
1479 ProjectsEvent::Verified { .. } => "verified",
1480 ProjectsEvent::RecordWritten => "record_written",
1481 ProjectsEvent::RecordFailed { .. } => "record_failed",
1482 ProjectsEvent::ChildrenSettling { .. } => "children_settling",
1483 ProjectsEvent::ChildChecked { .. } => "child_checked",
1484 ProjectsEvent::Finished { .. } => "finished",
1485 }
1486 }
1487
1488 fn fields(&self) -> serde_json::Value {
1489 match self {
1490 ProjectsEvent::DesignersPoll { open } => serde_json::json!({ "open": open }),
1491 ProjectsEvent::DesignersProceed { sessions } => {
1492 serde_json::json!({ "sessions": sessions_json(sessions) })
1493 }
1494 ProjectsEvent::DesignersSkipped => serde_json::json!({ "reason": "record only" }),
1495 ProjectsEvent::RunStarted { run_id, action } => {
1496 serde_json::json!({ "run_id": run_id, "action": action })
1497 }
1498 ProjectsEvent::PreImageExported { bytes } => serde_json::json!({ "bytes": bytes }),
1499 ProjectsEvent::ImportSent { overwrite } => {
1500 serde_json::json!({ "overwrite": overwrite })
1501 }
1502 ProjectsEvent::Verified { equal } => serde_json::json!({ "equal": equal }),
1503 ProjectsEvent::RecordFailed { code } => serde_json::json!({ "code": code }),
1504 ProjectsEvent::ChildrenSettling { children } => {
1505 serde_json::json!({ "children": children })
1506 }
1507 ProjectsEvent::ChildChecked { name, status } => {
1508 serde_json::json!({ "name": name, "status": status.as_str() })
1509 }
1510 ProjectsEvent::Finished { outcome } => {
1511 serde_json::json!({ "outcome": outcome.as_str() })
1512 }
1513 ProjectsEvent::ImportAnswered | ProjectsEvent::RecordWritten => serde_json::Value::Null,
1514 }
1515 }
1516}
1517
1518pub trait ProjectsObserver: Send + Sync {
1521 fn event(&self, event: &ProjectsEvent);
1523}
1524
1525pub struct NoProjectsObserver;
1527
1528impl ProjectsObserver for NoProjectsObserver {
1529 fn event(&self, _event: &ProjectsEvent) {}
1530}
1531
1532pub struct ProjectsRunOptions<'a> {
1534 pub repo: PathBuf,
1536 pub targets_root: PathBuf,
1539 pub target_file: String,
1541 pub master: NodeTarget,
1545 #[doc(hidden)]
1548 pub record_writer: Option<&'a dyn ReleaseRecordWriter>,
1549 pub designers: DesignersPolicy,
1551 pub children_settle: Duration,
1554 pub children_poll: Duration,
1558}
1559
1560struct RunEvents<'a> {
1564 log: EventLog,
1565 observer: &'a dyn ProjectsObserver,
1566 late_failure: bool,
1567}
1568
1569impl RunEvents<'_> {
1570 fn emit(&mut self, event: ProjectsEvent) -> Result<(), CoreError> {
1571 self.observer.event(&event);
1572 self.log.append(event.name(), event.fields())
1573 }
1574
1575 fn record(&mut self, event: &ProjectsEvent) -> Result<(), CoreError> {
1577 self.log.append(event.name(), event.fields())
1578 }
1579
1580 fn emit_late(&mut self, event: ProjectsEvent) {
1581 if self.emit(event).is_err() {
1582 self.late_failure = true;
1583 }
1584 }
1585}
1586
1587fn sessions_json(sessions: &[DesignerSeen]) -> Vec<serde_json::Value> {
1590 sessions
1591 .iter()
1592 .map(|(user, project, host)| {
1593 serde_json::json!({ "user": user, "project": project, "host": host })
1594 })
1595 .collect()
1596}
1597
1598struct DesignersFound {
1600 mode: &'static str,
1603 timeout_s: Option<u64>,
1605 polls: Vec<usize>,
1607 seen: Vec<DesignerSeen>,
1609 proceeded: bool,
1611 skipped: bool,
1613}
1614
1615async fn designers_check(
1620 plan: &DeployProjectsResult,
1621 gateway: &dyn ProjectGateway,
1622 policy: &DesignersPolicy,
1623 observer: &dyn ProjectsObserver,
1624) -> Result<DesignersFound, CoreError> {
1625 if plan.action == ProjectAction::Record {
1626 return Ok(DesignersFound {
1627 mode: "skipped",
1628 timeout_s: None,
1629 polls: Vec::new(),
1630 seen: Vec::new(),
1631 proceeded: false,
1632 skipped: true,
1633 });
1634 }
1635 let mut polls = Vec::new();
1636 let (verdict, seen) =
1637 designers_gate(|| gateway.designers(), policy, &SystemClock, &mut |open| {
1638 polls.push(open);
1639 observer.event(&ProjectsEvent::DesignersPoll { open });
1640 Ok(())
1641 })
1642 .await?;
1643 if verdict == DesignersVerdict::Open {
1644 return Err(CoreError::DeployPreflightFailed {
1645 failures: Box::new([PreflightFailure {
1646 check: "designers".to_string(),
1647 detail: designers_detail(&seen),
1648 }]),
1649 });
1650 }
1651 Ok(DesignersFound {
1652 mode: policy.mode.as_str(),
1653 timeout_s: Some(policy.timeout.as_secs()),
1654 polls,
1655 seen,
1656 proceeded: verdict == DesignersVerdict::Proceeded,
1657 skipped: false,
1658 })
1659}
1660
1661async fn wrong_target_check(
1666 loaded: &LoadedProject,
1667 gateway: &dyn ProjectGateway,
1668 opts: &ProjectsRunOptions<'_>,
1669) -> Result<(), CoreError> {
1670 let rest = gateway.gateway_name().await?;
1671 let reader = for_node(
1672 &opts.targets_root,
1673 &opts.target_file,
1674 "nodes.master",
1675 &opts.master,
1676 )?;
1677 let problems = wrong_target_gate(
1678 rest.as_deref(),
1679 reader.as_ref(),
1680 &loaded.target.mode,
1681 opts.master.gateway_name.as_deref(),
1682 )
1683 .await?;
1684 if problems.is_empty() {
1685 return Ok(());
1686 }
1687 Err(CoreError::DeployPreflightFailed {
1688 failures: Box::new([wrong_target_failure(&problems)]),
1689 })
1690}
1691
1692fn release_record(plan: &DeployProjectsResult, release: &str, run_id: &RunId) -> ReleaseRecord {
1694 ReleaseRecord {
1695 schema_version: 1,
1696 gateway: plan.gateway.clone(),
1697 project: plan.project.clone(),
1698 release: release.to_string(),
1699 artifact_sha256: plan.artifact_sha256.clone(),
1700 run_id: run_id.as_str().to_string(),
1701 deployed_at: rfc3339_now_utc(),
1702 }
1703}
1704
1705fn stale_live(expected: String, actual: String) -> CoreError {
1706 CoreError::PlanStale {
1707 expected,
1708 actual,
1709 detail: "live project changed since the plan".to_string(),
1710 }
1711}
1712
1713pub async fn deploy_projects_run(
1720 prepared: ProjectsPrepared,
1721 loaded: &LoadedProject,
1722 input: &ProjectsInput,
1723 surfaces: &ProjectsSurfaces<'_>,
1724 opts: &ProjectsRunOptions<'_>,
1725 observer: &dyn ProjectsObserver,
1726) -> Result<DeployProjectsResult, CoreError> {
1727 let plan = match prepared {
1728 ProjectsPrepared::Unchanged(result) => return Ok(result),
1729 ProjectsPrepared::Ready { plan, .. } => plan,
1730 };
1731 let built;
1734 let writer: &dyn ReleaseRecordWriter = match opts.record_writer {
1735 Some(writer) => writer,
1736 None => {
1737 built = release_writer_for_node(
1738 &opts.targets_root,
1739 &opts.target_file,
1740 "nodes.master",
1741 &opts.master,
1742 )?;
1743 built.as_ref()
1744 }
1745 };
1746 wrong_target_check(loaded, surfaces.gateway, opts).await?;
1749 let designers = designers_check(&plan, surfaces.gateway, &opts.designers, observer).await?;
1750 let (run_id, run_dir) = RunDir::create_fresh(&opts.repo, &plan.gateway)?;
1751 let rel = run_dir_rel(&plan.gateway, &run_id);
1752 let outcome = execute(
1753 &plan, loaded, input, surfaces, opts, writer, &run_dir, &run_id, &rel, &designers, observer,
1754 )
1755 .await;
1756 match outcome {
1757 Ok(mut result) => {
1758 if let Err(err) = run_dir.write_json("result.json", &result) {
1759 result.warnings.push(format!(
1760 "the run's result.json could not be written ({})",
1761 err.code()
1762 ));
1763 }
1764 Ok(result)
1765 }
1766 Err(err) => {
1767 let _ = run_dir.write_json(
1769 "result.json",
1770 &serde_json::json!({
1771 "ok": false,
1772 "slug": err.code(),
1773 "run_id": run_id.as_str(),
1774 }),
1775 );
1776 Err(err)
1777 }
1778 }
1779}
1780
1781struct ChildSweep {
1783 children: Vec<ChildReport>,
1784 warnings: Vec<String>,
1785}
1786
1787type ChildRead = Result<String, String>;
1790
1791async fn read_children(gateway: &dyn ProjectGateway, names: &[String]) -> Vec<ChildRead> {
1792 let mut reads = Vec::with_capacity(names.len());
1793 for name in names {
1794 reads.push(match gateway.export_bytes(name).await {
1795 Ok(zip) => live_digest(name, &zip)
1796 .map(|digest| digest.sha256)
1797 .map_err(|err| err.code().to_string()),
1798 Err(err) => Err(err.code().to_string()),
1799 });
1800 }
1801 reads
1802}
1803
1804async fn sweep_children(
1815 result: &DeployProjectsResult,
1816 loaded: &LoadedProject,
1817 surfaces: &ProjectsSurfaces<'_>,
1818 opts: &ProjectsRunOptions<'_>,
1819 events: &mut RunEvents<'_>,
1820) -> ChildSweep {
1821 let project = result.project.as_str();
1822 let gateway = surfaces.gateway;
1823 let mut sweep = ChildSweep {
1824 children: Vec::new(),
1825 warnings: Vec::new(),
1826 };
1827
1828 let records = match gateway.projects().await {
1829 Ok(records) => records,
1830 Err(err) => {
1831 sweep.warnings.push(format!(
1832 "could not list the projects to check the children of {project} ({})",
1833 err.code()
1834 ));
1835 return sweep;
1836 }
1837 };
1838 let mut names: Vec<String> = records
1839 .into_iter()
1840 .filter(|record| record.parent.as_deref() == Some(project))
1841 .map(|record| record.name)
1842 .collect();
1843 names.sort_by(|a, b| a.as_bytes().cmp(b.as_bytes()));
1844 names.dedup();
1845 if names.is_empty() {
1846 return sweep;
1847 }
1848 events.emit_late(ProjectsEvent::ChildrenSettling {
1849 children: names.clone(),
1850 });
1851
1852 let started = Instant::now();
1855 let mut current = read_children(gateway, &names).await;
1856 let mut unsettled = vec![true; names.len()];
1857 loop {
1858 let elapsed = started.elapsed();
1859 if elapsed >= opts.children_settle {
1860 break;
1861 }
1862 tokio::time::sleep(opts.children_poll.min(opts.children_settle - elapsed)).await;
1863 let next = read_children(gateway, &names).await;
1864 unsettled = current.iter().zip(&next).map(|(a, b)| a != b).collect();
1865 current = next;
1866 if !unsettled.iter().any(|changing| *changing) {
1867 break;
1868 }
1869 }
1870
1871 let gateway_name = result.gateway.as_str();
1872 let root = LedgerRoot::for_node(&loaded.target.nodes.master);
1873 for ((name, read), changing) in names.iter().zip(current).zip(unsettled) {
1874 let live = match read {
1875 Ok(sha256) => Some(sha256),
1876 Err(code) => {
1877 sweep.warnings.push(format!(
1878 "child project {name}: its export could not be read ({code})"
1879 ));
1880 None
1881 }
1882 };
1883 if changing {
1884 sweep.warnings.push(format!(
1885 "child project {name} did not settle within {} s",
1886 opts.children_settle.as_secs()
1887 ));
1888 }
1889 let record = match surfaces.host.read_release_record(gateway_name, name).await {
1890 Ok(None) => None,
1891 Ok(Some(bytes)) => {
1892 let location = release_record_location(root, gateway_name, name);
1893 match ReleaseRecord::parse(&bytes, &location, gateway_name, name) {
1894 Ok(record) => Some(record),
1895 Err(err) => {
1896 sweep.warnings.push(format!(
1897 "child project {name}: its release record could not be read ({})",
1898 err.code()
1899 ));
1900 None
1901 }
1902 }
1903 }
1904 Err(err) => {
1905 sweep.warnings.push(format!(
1906 "child project {name}: its release record could not be read ({})",
1907 err.code()
1908 ));
1909 None
1910 }
1911 };
1912 let status = match (&record, &live) {
1913 (Some(record), Some(live)) if record.artifact_sha256 == *live => ChildStatus::Ok,
1914 (Some(record), Some(_)) => {
1915 sweep.warnings.push(format!(
1916 "child project {name} no longer matches the release this tool last deployed to it ({})",
1917 record.release
1918 ));
1919 ChildStatus::Mismatch
1920 }
1921 _ => ChildStatus::Unverified,
1922 };
1923 events.emit_late(ProjectsEvent::ChildChecked {
1924 name: name.clone(),
1925 status,
1926 });
1927 sweep.children.push(ChildReport {
1928 name: name.clone(),
1929 status,
1930 release: record.as_ref().map(|r| r.release.clone()),
1931 recorded_sha256: record.map(|r| r.artifact_sha256),
1932 live_sha256: live,
1933 });
1934 }
1935 sweep
1936}
1937
1938#[allow(clippy::too_many_arguments)]
1939async fn execute(
1940 plan: &DeployProjectsResult,
1941 loaded: &LoadedProject,
1942 input: &ProjectsInput,
1943 surfaces: &ProjectsSurfaces<'_>,
1944 opts: &ProjectsRunOptions<'_>,
1945 writer: &dyn ReleaseRecordWriter,
1946 run_dir: &RunDir,
1947 run_id: &RunId,
1948 rel: &str,
1949 designers: &DesignersFound,
1950 observer: &dyn ProjectsObserver,
1951) -> Result<DeployProjectsResult, CoreError> {
1952 let project = plan.project.as_str();
1953 let gateway = surfaces.gateway;
1954 let artifact = &loaded.artifact.digest;
1955
1956 run_dir.write_json(
1957 "run.json",
1958 &serde_json::json!({
1959 "run_id": run_id.as_str(),
1960 "started_at": rfc3339_now_utc(),
1961 "env": plan.env,
1962 "gateway": plan.gateway,
1963 "project": plan.project,
1964 "release": plan.release,
1965 "action": plan.action,
1966 "artifact_sha256": plan.artifact_sha256,
1967 "live_sha256": plan.live_sha256,
1968 "allow_downgrade": input.allow_downgrade,
1969 "allow_first_change": input.allow_first_change,
1970 "designers": designers.mode,
1971 "designers_timeout_s": designers.timeout_s,
1972 "designers_seen": sessions_json(&designers.seen),
1973 "run_dir_git": run_dir.git_status().as_record(),
1974 }),
1975 )?;
1976 run_dir.write_json("plan.json", plan)?;
1977 let mut events = RunEvents {
1978 log: EventLog::open(run_dir)?,
1979 observer,
1980 late_failure: false,
1981 };
1982 events.emit(ProjectsEvent::RunStarted {
1983 run_id: run_id.as_str().to_string(),
1984 action: plan.action,
1985 })?;
1986 for open in &designers.polls {
1989 events.record(&ProjectsEvent::DesignersPoll { open: *open })?;
1990 }
1991 if designers.proceeded {
1992 events.emit(ProjectsEvent::DesignersProceed {
1993 sessions: designers.seen.clone(),
1994 })?;
1995 }
1996 if designers.skipped {
1997 events.emit(ProjectsEvent::DesignersSkipped)?;
1998 }
1999
2000 let mut result = plan.clone();
2001 result.run_id = Some(run_id.as_str().to_string());
2002 result.run_dir = Some(rel.to_string());
2003 let record = release_record(plan, &loaded.release.tag, run_id);
2004
2005 if plan.action == ProjectAction::Record {
2006 let zip = gateway.export_bytes(project).await?;
2009 let live = live_digest(project, &zip)?;
2010 if live.sha256 != artifact.sha256 {
2011 return Err(stale_live(artifact.sha256.clone(), live.sha256));
2012 }
2013 events.emit(ProjectsEvent::Verified { equal: true })?;
2014 if let Err(err) = writer
2015 .write_release_record(&plan.gateway, project, &record.to_bytes())
2016 .await
2017 {
2018 events.emit_late(ProjectsEvent::RecordFailed {
2019 code: err.code().to_string(),
2020 });
2021 return Err(err);
2022 }
2023 finish_recorded(&mut result, &mut events);
2027 return Ok(result);
2028 }
2029
2030 let overwrite = plan.action == ProjectAction::Import;
2031 if overwrite {
2032 let pre = gateway.export_bytes(project).await?;
2036 let live = live_digest(project, &pre)?;
2037 let mut file = run_dir.create_private_file_in("export", &format!("{project}.zip"))?;
2038 file.write_all(&pre)
2039 .and_then(|()| file.sync_all())
2040 .map_err(|e| CoreError::DeployRunInvalid {
2041 run_id: run_id.as_str().to_string(),
2042 problem: format!("could not write the pre-image export ({:?})", e.kind()),
2043 })?;
2044 events.emit(ProjectsEvent::PreImageExported { bytes: pre.len() })?;
2045 if plan.live_sha256.as_deref() != Some(live.sha256.as_str()) {
2048 return Err(stale_live(
2049 plan.live_sha256
2050 .clone()
2051 .unwrap_or_else(|| "none".to_string()),
2052 live.sha256,
2053 ));
2054 }
2055 }
2056
2057 events.emit(ProjectsEvent::ImportSent { overwrite })?;
2058 let answer = match gateway
2059 .project_import(project, loaded.artifact.zip.clone(), overwrite)
2060 .await
2061 {
2062 Ok(answer) => Some(answer),
2063 Err(err) if import_outcome_unknown(&err) => {
2068 result.warnings.push(format!(
2069 "the import request failed after it may have reached the gateway ({}); the project was read back to find out what it holds",
2070 err.code()
2071 ));
2072 None
2073 }
2074 Err(err) => return Err(err),
2076 };
2077 if let Some(answer) = &answer {
2080 events.emit_late(ProjectsEvent::ImportAnswered);
2081 if answer
2082 .response
2083 .get("changes")
2084 .and_then(serde_json::Value::as_array)
2085 .is_some_and(Vec::is_empty)
2086 {
2087 result
2088 .warnings
2089 .push("the gateway reported no changes for this import".to_string());
2090 }
2091 }
2092
2093 let equal = match gateway.export_bytes(project).await {
2094 Ok(zip) => match live_digest(project, &zip) {
2095 Ok(live) => Some(live.sha256 == artifact.sha256),
2096 Err(err) => {
2097 result.warnings.push(format!(
2098 "could not verify the import: the project read back is unusable ({})",
2099 err.code()
2100 ));
2101 None
2102 }
2103 },
2104 Err(err) => {
2105 result.warnings.push(format!(
2106 "could not verify the import: the project could not be read back ({})",
2107 err.code()
2108 ));
2109 None
2110 }
2111 };
2112 events.emit_late(ProjectsEvent::Verified {
2113 equal: equal == Some(true),
2114 });
2115
2116 if equal == Some(true) {
2117 match writer
2118 .write_release_record(&plan.gateway, project, &record.to_bytes())
2119 .await
2120 {
2121 Ok(()) => {
2122 events.emit_late(ProjectsEvent::RecordWritten);
2123 result.record_written = true;
2124 }
2125 Err(err) => {
2126 events.emit_late(ProjectsEvent::RecordFailed {
2127 code: err.code().to_string(),
2128 });
2129 result.warnings.push(format!(
2130 "the import is verified but the release record could not be written ({})",
2131 err.code()
2132 ));
2133 }
2134 }
2135 if let Some(parent) = &loaded.meta.parent {
2138 match gateway.projects().await {
2139 Ok(records) => {
2140 if records
2141 .iter()
2142 .find(|r| r.name == project)
2143 .is_some_and(|r| r.invalid_parent == Some(true))
2144 {
2145 result.warnings.push(format!(
2146 "project {project}: after the import the gateway reports that its parent {parent} is invalid"
2147 ));
2148 }
2149 }
2150 Err(err) => result.warnings.push(format!(
2151 "project {project}: the parent could not be read back after the import ({})",
2152 err.code()
2153 )),
2154 }
2155 }
2156 result.outcome = if overwrite {
2157 ProjectOutcome::Imported
2158 } else {
2159 ProjectOutcome::Created
2160 };
2161 } else {
2162 result.outcome = ProjectOutcome::ImportedUnverified;
2165 result.ok = false;
2166 result.slug = Some(PROJECT_UNVERIFIED_SLUG.to_string());
2167 if overwrite {
2171 result.restore_command = Some(restore_command(project, rel));
2172 }
2173 }
2174 if matches!(
2178 result.outcome,
2179 ProjectOutcome::Imported | ProjectOutcome::ImportedUnverified
2180 ) {
2181 let sweep = sweep_children(&result, loaded, surfaces, opts, &mut events).await;
2182 result.children = sweep.children;
2183 result.warnings.extend(sweep.warnings);
2184 }
2185 events.emit_late(ProjectsEvent::Finished {
2186 outcome: result.outcome,
2187 });
2188 note_late_log_failure(&mut result, &events);
2189 Ok(result)
2190}
2191
2192fn import_outcome_unknown(err: &CoreError) -> bool {
2199 match err {
2200 CoreError::Network {
2201 source: Some(source),
2202 ..
2203 } => !(source.is_connect() || source.is_builder()),
2204 CoreError::Network { source: None, .. } | CoreError::Internal(_) => true,
2205 _ => false,
2206 }
2207}
2208
2209fn finish_recorded(result: &mut DeployProjectsResult, events: &mut RunEvents<'_>) {
2213 result.record_written = true;
2214 result.outcome = ProjectOutcome::Recorded;
2215 events.emit_late(ProjectsEvent::RecordWritten);
2216 events.emit_late(ProjectsEvent::Finished {
2217 outcome: result.outcome,
2218 });
2219 note_late_log_failure(result, events);
2220}
2221
2222fn note_late_log_failure(result: &mut DeployProjectsResult, events: &RunEvents<'_>) {
2224 if events.late_failure {
2225 result
2226 .warnings
2227 .push("the run's events log could not be written to the end".to_string());
2228 }
2229}
2230
2231#[cfg(test)]
2232mod tests {
2233 use super::*;
2234
2235 fn sample_zip() -> Vec<u8> {
2236 use std::io::Write as _;
2237 let mut writer = zip::ZipWriter::new(std::io::Cursor::new(Vec::new()));
2238 let options = zip::write::SimpleFileOptions::default()
2239 .compression_method(zip::CompressionMethod::Deflated);
2240 for (name, bytes) in [
2241 ("project.json", br#"{"title":"T"}"#.as_slice()),
2242 ("ignition/s/resource.json", br#"{"files":["a.py"]}"#),
2243 ("ignition/s/a.py", &[b'x'; 4096]),
2244 ] {
2245 writer.start_file(name, options).expect("start");
2246 writer.write_all(bytes).expect("write");
2247 }
2248 writer.finish().expect("finish").into_inner()
2249 }
2250
2251 #[test]
2252 fn check_artifact_zip_refuses_truncated() {
2253 let whole = sample_zip();
2254 check_artifact_zip(&whole).expect("the whole zip passes");
2255 let half = &whole[..whole.len() / 2];
2256 let err = check_artifact_zip(half).expect_err("half a zip");
2257 assert_eq!(err.code(), "invalid_import_file");
2258 let err = check_artifact_zip(&whole[..whole.len() - 40]).expect_err("tail cut");
2260 assert_eq!(err.code(), "invalid_import_file");
2261 let err = check_artifact_zip(b"not a zip").expect_err("no magic");
2262 assert_eq!(err.code(), "invalid_import_file");
2263 }
2264
2265 fn sample_result() -> DeployProjectsResult {
2266 DeployProjectsResult {
2267 env: "prod".to_string(),
2268 gateway: "gw".to_string(),
2269 project: "CapApp".to_string(),
2270 release: "v1.0.0".to_string(),
2271 check: false,
2272 action: ProjectAction::Record,
2273 would_change: true,
2274 artifact_sha256: "a".repeat(64),
2275 live_sha256: Some("a".repeat(64)),
2276 members: MemberDelta::default(),
2277 removed: Vec::new(),
2278 removed_count: 0,
2279 excluded: Vec::new(),
2280 excluded_count: 0,
2281 unmanaged: Vec::new(),
2282 unmanaged_count: 0,
2283 parent: None,
2284 previous_release: None,
2285 recorded_sha256: None,
2286 live_matches_last_deploy: None,
2287 allow_downgrade: false,
2288 allow_first_change: None,
2289 warnings: Vec::new(),
2290 run_id: None,
2291 run_dir: None,
2292 record_written: false,
2293 restore_command: None,
2294 children: Vec::new(),
2295 missing_tags: Some(Vec::new()),
2296 missing_tags_count: Some(0),
2297 missing_tags_unknown_providers: Some(Vec::new()),
2298 missing_tags_unknown_providers_count: Some(0),
2299 missing_tags_unavailable: None,
2300 missing_tags_limit: MISSING_TAGS_LIMIT,
2301 outcome: ProjectOutcome::Checked,
2302 ok: true,
2303 slug: None,
2304 }
2305 }
2306
2307 #[test]
2311 fn a_log_failure_after_the_record_is_written_is_a_warning() {
2312 let dir = tempfile::tempdir().expect("tempdir");
2313 let path = dir.path().join("events.ndjson");
2314 std::fs::write(&path, b"").expect("seed");
2315 let read_only = std::fs::File::open(&path).expect("open read-only");
2317 let mut events = RunEvents {
2318 log: EventLog::from_file(read_only),
2319 observer: &NoProjectsObserver,
2320 late_failure: false,
2321 };
2322 let mut result = sample_result();
2323
2324 finish_recorded(&mut result, &mut events);
2325
2326 assert!(result.record_written, "the record is on the host");
2327 assert_eq!(result.outcome, ProjectOutcome::Recorded);
2328 assert!(result.ok, "a log failure does not fail the run");
2329 assert!(events.late_failure);
2330 assert!(
2331 result
2332 .warnings
2333 .iter()
2334 .any(|w| w.contains("events log could not be written")),
2335 "{:?}",
2336 result.warnings
2337 );
2338 }
2339
2340 #[test]
2341 fn a_built_artifact_passes_the_check() {
2342 let dir = tempfile::tempdir().expect("tempdir");
2343 for (rel, bytes) in [
2344 ("project.json", br#"{"title":"T"}"#.as_slice()),
2345 ("ignition/s/resource.json", br#"{"files":["a.py"]}"#),
2346 ("ignition/s/a.py", b"pass"),
2347 ] {
2348 let path = dir.path().join(rel);
2349 std::fs::create_dir_all(path.parent().expect("parent")).expect("mkdir");
2350 std::fs::write(path, bytes).expect("write");
2351 }
2352 let artifact = build_artifact(dir.path()).expect("artifact");
2353 check_artifact_zip(&artifact.zip).expect("a built artifact is a valid import");
2354 }
2355}