1use std::collections::{BTreeMap, BTreeSet, VecDeque};
226use std::io::Write as _;
227use std::path::PathBuf;
228use std::time::{Duration, Instant};
229
230use async_trait::async_trait;
231use serde::Serialize;
232use serde_json::Value;
233
234use super::deploy::load_target;
235use super::projects::validate_import;
236use super::resources::export_zip_bytes;
237use super::workspace::rfc3339_now_utc;
238use crate::client::deploy_api::scan_and_wait;
239use crate::client::projects::{ImportOutcome, ProjectRecord};
240use crate::client::query::ListQuery;
241use crate::client::scan::{LockVerdict, ScanKind, ScanOutcome};
242use crate::client::tags::{
243 CollisionPolicy, TagImportAnswer, TagProviderCreate, TagProviderModify, TagProviderRecord,
244};
245use crate::client::{DeployApi, GatewayApi};
246use crate::deploy::apply::fingerprint::fingerprint;
247use crate::deploy::apply::hold::SystemClock;
248use crate::deploy::apply::hold::{HOLD_S, HeldLock, LOCK_MARGIN_S};
249use crate::deploy::apply::host::{ApplyHost, BatchSpec, DeleteOp, WriteOp};
250use crate::deploy::apply::journal::{Journal, JournalPhase};
251use crate::deploy::apply::preflight::{
252 DesignerSeen, DesignersPolicy, DesignersVerdict, designers_detail, designers_gate,
253 read_designers, wrong_target_failure, wrong_target_gate,
254};
255use crate::deploy::apply::record::{RecordState, RunRecord, TouchedEntry};
256use crate::deploy::apply::rollback::live_changes;
257use crate::deploy::apply::run_store::{EventLog, RunDir, RunId, run_dir_rel};
258use crate::deploy::apply::{BACKUP_UNSYNCED_SLUG, ROLLED_BACK_SLUG};
259use crate::deploy::exclude;
260use crate::deploy::ledger::{
261 LedgerRoot, journal_location, pre_image_location, project_journal_location,
262 release_record_location, tag_record_location,
263};
264use crate::deploy::lint::{Sends, scan_tag_node};
265use crate::deploy::manifest::Manifest;
266use crate::deploy::native_tags::{
267 NativeNode, NativeProvider, NonCanonical, TAG_GROUPS_FILE, check_canonical_form,
268 read_provider_with_raw,
269};
270use crate::deploy::pair::{BackupPolicy, wall_clock_ms};
271use crate::deploy::project_artifact::{
272 Artifact, CanonicalDigest, build_artifact, canonical_digest,
273};
274use crate::deploy::project_run::{
275 CreatedProof, JournalState, PreImageFile, PreImageProvider, PreImageRun, ProjectJournal,
276};
277use crate::deploy::project_tags::{
278 FILES_PER_ENTRY, Lookup, MISSING_TAGS_LIMIT, MissingTag, TagRef, TagRefs, TagTree,
279 TagsUnavailable, UnknownProvider, scan_artifact,
280};
281use crate::deploy::projects::{
282 MemberDelta, PROJECT_FILE_REL, ParentLink, ProjectAction, ProjectFile, ProjectMeta,
283 ProviderSettings, ReadOnlySources, Release, ReleaseOrder, ReleaseRecord, check_parent,
284 check_tag_ownership, gateway_manifest_rel, load_project_file, load_provider_settings,
285 member_delta, parent_chain, read_project_meta,
286};
287use crate::deploy::tag_apply::{
288 AppliedSteps, Carried, ImportBatch, REMOVED_PROPERTY_POLICY, TAG_DELETE_DESCENDANTS, carried,
289 created_deletes_now, delete_calls, import_batches, rollback_over_cap, rollback_plan,
290 uncarriable_in_batch,
291};
292use crate::deploy::tag_canon::{CanonOpts, canonical, diff_properties, flatten_export, to_node};
293use crate::deploy::tag_converge::{
294 BackupTags, DeployResync, MasterResync, TagsBackupEvent, TagsBackupReport, converge_tags,
295};
296use crate::deploy::tag_delete::{RouteHandshake, TagDeleteRoute, delete_chunks, wait_ready};
297use crate::deploy::tag_groups::{
298 GroupPlan, GroupSummary, LiveGroups, check_group_refs, live_group_users, parse_group_value,
299 plan_groups, referenced_groups,
300};
301use crate::deploy::tag_plan::{
302 NodeChange, ProviderPlan, ProviderSettingsAction, ProviderSettingsPlan, TagDeleteRouteReport,
303 TagJournal, TagPlan, TagRollback, TagVerify, TagsReport, plan_provider_settings, plan_tags,
304 removed_types_in_use, repo_tags_sha256,
305};
306use crate::deploy::tag_record::TagRecord;
307use crate::deploy::target::{NodeTarget, Target};
308use crate::deploy::transport::projects::{
309 ProjectHost, ProjectRunWriter, ReleaseRecordWriter, TagRecordWriter, git_entry_location,
310 project_run_writer_for_node, release_writer_for_node, tag_record_names_problem,
311 tag_record_writer_for_node,
312};
313use crate::deploy::transport::remote::safe_remote_path;
314use crate::deploy::transport::{TreeReader, apply_host_for_node, for_node};
315use crate::deploy::tree::read_repo_file;
316use crate::deploy::tree::{RenderedTree, sha256_hex};
317use crate::error::{
318 APPLY_LIST_SHOWN, CoreError, DeployFinding, PreflightFailure, RollbackFailure, RollbackScope,
319};
320
321pub const LIST_CAP: usize = 200;
323
324pub const CHILDREN_SETTLE_S: u64 = 30;
328
329pub const CHILDREN_POLL_S: u64 = 5;
331
332pub const TAG_VERIFY_SETTLE_S: u64 = 30;
338
339pub const TAG_VERIFY_POLL_S: u64 = 5;
342
343pub const PROJECTS_CANARY_EXEMPT: &[&str] = &["export"];
353
354#[async_trait]
364pub trait ProjectGateway: Send + Sync {
365 async fn projects(&self) -> Result<Vec<ProjectRecord>, CoreError>;
367 async fn project_find(&self, name: &str) -> Result<Option<ProjectRecord>, CoreError>;
369 async fn export_bytes(&self, name: &str) -> Result<Vec<u8>, CoreError>;
371 async fn designers(&self) -> Result<Vec<DesignerSeen>, CoreError>;
374 async fn gateway_name(&self) -> Result<Option<String>, CoreError>;
378 async fn tag_provider_names(&self) -> Result<Vec<String>, CoreError>;
381 async fn tag_export(&self, provider: &str) -> Result<Vec<u8>, CoreError>;
384 async fn project_import(
388 &self,
389 name: &str,
390 zip: Vec<u8>,
391 overwrite: bool,
392 ) -> Result<ImportOutcome, CoreError>;
393 async fn tag_import(
399 &self,
400 provider: &str,
401 path: &str,
402 policy: CollisionPolicy,
403 body: Vec<u8>,
404 ) -> Result<TagImportAnswer, CoreError>;
405 async fn tag_provider_record(&self, name: &str)
408 -> Result<Option<TagProviderRecord>, CoreError>;
409 async fn tag_provider_create(&self, body: &[TagProviderCreate]) -> Result<(), CoreError>;
412 async fn tag_provider_modify(&self, body: &[TagProviderModify]) -> Result<(), CoreError>;
416 async fn tag_provider_delete(&self, name: &str, signature: &str) -> Result<(), CoreError>;
420}
421
422pub struct GatewayProjects<'a>(pub &'a dyn GatewayApi);
424
425#[async_trait]
426impl ProjectGateway for GatewayProjects<'_> {
427 async fn projects(&self) -> Result<Vec<ProjectRecord>, CoreError> {
428 Ok(self.0.projects(&ListQuery::default()).await?.items)
429 }
430
431 async fn project_find(&self, name: &str) -> Result<Option<ProjectRecord>, CoreError> {
432 match self.0.project_find(name).await {
433 Ok(record) => Ok(Some(record)),
434 Err(CoreError::NotFound { .. }) => Ok(None),
435 Err(err) => Err(err),
436 }
437 }
438
439 async fn export_bytes(&self, name: &str) -> Result<Vec<u8>, CoreError> {
440 export_zip_bytes(self.0, name).await
441 }
442
443 async fn designers(&self) -> Result<Vec<DesignerSeen>, CoreError> {
444 read_designers(self.0).await
445 }
446
447 async fn gateway_name(&self) -> Result<Option<String>, CoreError> {
448 Ok(self
449 .0
450 .gateway_info()
451 .await?
452 .name
453 .filter(|name| !name.trim().is_empty()))
454 }
455
456 async fn tag_provider_names(&self) -> Result<Vec<String>, CoreError> {
457 Ok(self
458 .0
459 .tag_provider_list(&ListQuery::default())
460 .await?
461 .items
462 .into_iter()
463 .map(|record| record.name)
464 .collect())
465 }
466
467 async fn tag_export(&self, provider: &str) -> Result<Vec<u8>, CoreError> {
468 self.0.tags_export_provider(provider).await
469 }
470
471 async fn project_import(
472 &self,
473 name: &str,
474 zip: Vec<u8>,
475 overwrite: bool,
476 ) -> Result<ImportOutcome, CoreError> {
477 self.0.project_import(name, zip, overwrite).await
478 }
479
480 async fn tag_import(
481 &self,
482 provider: &str,
483 path: &str,
484 policy: CollisionPolicy,
485 body: Vec<u8>,
486 ) -> Result<TagImportAnswer, CoreError> {
487 self.0.tags_import(provider, path, policy, body).await
488 }
489
490 async fn tag_provider_record(
491 &self,
492 name: &str,
493 ) -> Result<Option<TagProviderRecord>, CoreError> {
494 match self.0.tag_provider_find(name).await {
495 Ok(record) => Ok(Some(record)),
496 Err(CoreError::NotFound { .. }) => Ok(None),
497 Err(err) => Err(err),
498 }
499 }
500
501 async fn tag_provider_create(&self, body: &[TagProviderCreate]) -> Result<(), CoreError> {
502 self.0.tag_provider_create(body).await
503 }
504
505 async fn tag_provider_modify(&self, body: &[TagProviderModify]) -> Result<(), CoreError> {
506 self.0.tag_provider_modify(body).await
507 }
508
509 async fn tag_provider_delete(&self, name: &str, signature: &str) -> Result<(), CoreError> {
510 self.0.tag_provider_delete(name, signature).await
511 }
512}
513
514pub struct ProjectsSurfaces<'a> {
519 pub gateway: &'a dyn ProjectGateway,
521 pub host: &'a dyn ProjectHost,
523 pub tree: Option<&'a dyn TreeReader>,
528 pub apply_host: Option<&'a dyn ApplyHost>,
533 pub deploy: Option<&'a dyn DeployApi>,
536 pub tag_delete: TagDeleteSurface<'a>,
540}
541
542#[derive(Clone)]
544pub enum TagDeleteSurface<'a> {
545 Configured(&'a dyn TagDeleteRoute),
547 NotConfigured {
550 missing: Vec<String>,
552 },
553}
554
555struct RouteState<'a> {
557 route: Option<&'a dyn TagDeleteRoute>,
559 missing: Vec<String>,
561 handshake: Option<RouteHandshake>,
563 problem: Option<String>,
565}
566
567impl<'a> RouteState<'a> {
568 fn of(
570 surface: &TagDeleteSurface<'a>,
571 answer: Option<Result<RouteHandshake, CoreError>>,
572 ) -> Self {
573 match surface {
574 TagDeleteSurface::Configured(route) => {
575 let (handshake, problem) = match answer {
576 Some(Ok(handshake)) => (Some(handshake), None),
577 Some(Err(err)) => (None, Some(err.code().to_string())),
578 None => (None, None),
579 };
580 Self {
581 route: Some(*route),
582 missing: Vec::new(),
583 handshake,
584 problem,
585 }
586 }
587 TagDeleteSurface::NotConfigured { missing } => Self {
588 route: None,
589 missing: missing.clone(),
590 handshake: None,
591 problem: None,
592 },
593 }
594 }
595
596 fn ready(&self) -> Option<(&'a dyn TagDeleteRoute, &RouteHandshake)> {
598 Some((self.route?, self.handshake.as_ref()?))
599 }
600
601 fn report(&self, owned: &[&str]) -> TagDeleteRouteReport {
603 let mut providers_not_allowed: Vec<String> = match &self.handshake {
604 Some(handshake) => owned
605 .iter()
606 .filter(|name| !handshake.providers.iter().any(|p| p == *name))
607 .map(|name| (*name).to_string())
608 .collect(),
609 None => Vec::new(),
610 };
611 providers_not_allowed.sort_by(|a, b| a.as_bytes().cmp(b.as_bytes()));
612 TagDeleteRouteReport {
613 configured: self.route.is_some(),
614 missing: self.missing.clone(),
615 ready: self.handshake.is_some(),
616 route_version: self.handshake.as_ref().map(|h| h.route_version.clone()),
617 providers_not_allowed,
618 problem: self.problem.clone(),
619 }
620 }
621
622 fn kept_reason(&self, provider: &str) -> String {
624 match (self.route, &self.handshake) {
625 (None, _) => "created by this run; this release has no tag delete".to_string(),
627 (Some(_), None) => format!(
628 "created by this run; the tag-delete route is not ready ({}), so it was not \
629 removed",
630 self.problem.as_deref().unwrap_or("no handshake")
631 ),
632 (Some(_), Some(handshake)) if !handshake.providers.iter().any(|p| p == provider) => {
633 format!(
634 "created by this run; the tag-delete route does not allow provider \
635 {provider}, so it was not removed"
636 )
637 }
638 (Some(_), Some(_)) => "created by this run; it holds a node this run did not \
639 create, so it was not removed"
640 .to_string(),
641 }
642 }
643}
644
645fn owned_names(loaded: &LoadedProject) -> Vec<&str> {
647 sorted(&loaded.owned_tags)
648 .into_iter()
649 .map(|p| p.name.as_str())
650 .collect()
651}
652
653#[derive(Debug, Clone)]
655pub struct ProjectsInput {
656 pub repo: PathBuf,
658 pub targets: PathBuf,
660 pub env: String,
662 pub release: String,
664 pub allow_downgrade: bool,
667 pub allow_first_change: Option<String>,
670 pub allow_destructive: bool,
674 pub artifact_sha256: Option<String>,
679 pub tags_sha256: Option<String>,
686}
687
688pub const FIRST_CHANGE_REASON_MAX: usize = 500;
690
691#[derive(Debug)]
693pub struct LoadedProject {
694 pub file: ProjectFile,
696 pub meta: ProjectMeta,
698 pub target_file: String,
700 pub target: Target,
702 pub artifact: Artifact,
704 pub release: Release,
706 pub owned_tags: Vec<NativeProvider>,
709 pub provider_settings: Vec<ProviderSettings>,
713 pub tag_exclusions: Vec<String>,
719 pub ownership_warnings: Vec<String>,
722 pub non_canonical: std::collections::BTreeMap<String, Vec<NonCanonical>>,
726}
727
728pub fn load_project(input: &ProjectsInput) -> Result<LoadedProject, CoreError> {
730 let release = Release::parse(&input.release)?;
733 load_with(input, release)
734}
735
736pub fn load_project_for_rollback(input: &ProjectsInput) -> Result<LoadedProject, CoreError> {
741 load_with(
742 input,
743 Release {
744 tag: String::new(),
745 version: semver::Version::new(0, 0, 0),
746 },
747 )
748}
749
750fn load_with(input: &ProjectsInput, release: Release) -> Result<LoadedProject, CoreError> {
751 if let Some(reason) = &input.allow_first_change {
752 let chars = reason.chars().count();
753 if chars == 0 || chars > FIRST_CHANGE_REASON_MAX {
754 return Err(CoreError::InvalidInput {
755 reason: format!(
756 "--allow-first-change <reason>: the reason is 1 to {FIRST_CHANGE_REASON_MAX} characters"
757 ),
758 });
759 }
760 }
761 if let Some(hash) = &input.artifact_sha256
762 && !is_lower_hex64(hash)
763 {
764 return Err(CoreError::InvalidInput {
765 reason: "--artifact-sha256 must be 64 lowercase hex characters (the \
766 `artifact_sha256` of `ign deploy projects --check --json`)"
767 .to_string(),
768 });
769 }
770 if let Some(hash) = &input.tags_sha256
771 && !is_lower_hex64(hash)
772 {
773 return Err(CoreError::InvalidInput {
774 reason: "--tags-sha256 must be 64 lowercase hex characters (the \
775 `data.tags.tags_sha256` of `ign deploy projects --check --json`)"
776 .to_string(),
777 });
778 }
779 let file = load_project_file(&input.repo)?;
780 file.require_env(&input.env)?;
781 let meta = read_project_meta(&input.repo)?;
782 let (target_file, target) = load_target(&input.targets, &input.env)?;
783 let (tag_exclusions, ownership_warnings) = if file.tag_providers.is_empty() {
788 (exclude::effective(&target, None), Vec::new())
789 } else {
790 let manifest_rel = gateway_manifest_rel(&target.gateway);
791 let manifest = read_gateway_manifest(&input.targets, &manifest_rel)?;
792 let manifest = manifest.as_ref().map(|m| (manifest_rel.as_str(), m));
793 let sources = ReadOnlySources::new(&target_file, &target, manifest);
794 let warnings = check_tag_ownership(&file, &sources, manifest)?;
795 (
796 exclude::effective(&target, manifest.map(|(_, m)| m)),
797 warnings,
798 )
799 };
800 record_names_fit_the_host(
803 &file.project,
804 &file.tag_providers,
805 &target.gateway,
806 &target_file,
807 &target.nodes.master,
808 )?;
809 let artifact = build_artifact(&input.repo)?;
810 check_artifact_zip(&artifact.zip)?;
811 if let Some(expected) = &input.artifact_sha256
815 && *expected != artifact.digest.sha256
816 {
817 return Err(CoreError::ProjectHashMismatch {
818 project: file.project.clone(),
819 hash: "artifact_sha256",
820 expected: expected.clone(),
821 actual: artifact.digest.sha256.clone(),
822 });
823 }
824 let provider_settings = file
827 .tag_providers
828 .iter()
829 .map(|provider| load_provider_settings(&input.repo, provider))
830 .collect::<Result<Vec<_>, _>>()?;
831 let mut owned_tags = Vec::new();
832 let mut non_canonical = std::collections::BTreeMap::new();
833 for provider in &file.tag_providers {
834 let (tags, raw) = read_provider_with_raw(&input.repo, provider)?;
835 if tags.nodes.is_empty() {
837 return Err(CoreError::deploy_tree_invalid(
838 format!("{}/{provider}", crate::deploy::native_tags::TAGS_DIR),
839 None,
840 None,
841 "holds no tag node: an owned provider is never deployed empty; drop it from \
842 tag_providers in deploy/project.yaml or keep at least one node"
843 .to_string(),
844 ));
845 }
846 non_canonical.insert(provider.clone(), check_canonical_form(&tags, &raw));
847 owned_tags.push(tags);
848 }
849 if let Some(expected) = &input.tags_sha256 {
852 if owned_tags.is_empty() {
853 return Err(CoreError::InvalidInput {
854 reason: "--tags-sha256 is given, but deploy/project.yaml lists no \
855 tag_providers: there is no owned tag content to compare it with"
856 .to_string(),
857 });
858 }
859 let actual = repo_tags_sha256(&owned_tags);
860 if *expected != actual {
861 return Err(CoreError::ProjectHashMismatch {
862 project: file.project.clone(),
863 hash: "tags_sha256",
864 expected: expected.clone(),
865 actual,
866 });
867 }
868 }
869 Ok(LoadedProject {
870 file,
871 meta,
872 target_file,
873 target,
874 artifact,
875 release,
876 owned_tags,
877 provider_settings,
878 tag_exclusions,
879 ownership_warnings,
880 non_canonical,
881 })
882}
883
884fn read_gateway_manifest(
888 targets: &std::path::Path,
889 rel: &str,
890) -> Result<Option<Manifest>, CoreError> {
891 match std::fs::symlink_metadata(targets.join(rel)) {
892 Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
893 _ => {}
894 }
895 let bytes = read_repo_file(targets, rel, None)?;
896 Manifest::parse(rel, &bytes).map(Some)
897}
898
899fn record_names_fit_the_host(
913 project: &str,
914 tag_providers: &[String],
915 gateway: &str,
916 target_file: &str,
917 master: &NodeTarget,
918) -> Result<(), CoreError> {
919 if master.ssh.is_some() {
920 if let Err(why) = safe_remote_path(project) {
921 return Err(CoreError::deploy_tree_invalid(
922 PROJECT_FILE_REL,
923 None,
924 None,
925 format!(
926 "project: the name cannot be used for the release record on an ssh node ({why})"
927 ),
928 ));
929 }
930 if let Err(why) = safe_remote_path(gateway) {
931 return Err(CoreError::deploy_tree_invalid(
932 target_file,
933 None,
934 None,
935 format!(
936 "gateway: the name cannot be used for the release record on an ssh node ({why})"
937 ),
938 ));
939 }
940 }
941 let findings: Vec<crate::error::DeployFinding> = tag_providers
942 .iter()
943 .enumerate()
944 .filter_map(|(i, provider)| {
945 tag_record_names_problem(gateway, provider, master).map(|why| {
946 crate::error::DeployFinding {
947 file: PROJECT_FILE_REL.to_string(),
948 line: None,
949 column: None,
950 message: format!(
951 "tag_providers[{i}]: the name cannot be used for the tag record on \
952 the master node ({why})"
953 ),
954 }
955 })
956 })
957 .collect();
958 if !findings.is_empty() {
959 return Err(CoreError::deploy_findings(findings));
960 }
961 Ok(())
962}
963
964fn is_lower_hex64(text: &str) -> bool {
966 text.len() == 64
967 && text
968 .bytes()
969 .all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b))
970}
971
972pub(crate) fn check_artifact_zip(zip: &[u8]) -> Result<(), CoreError> {
977 validate_import(zip)
978}
979
980#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
982#[serde(rename_all = "snake_case")]
983pub enum ProjectOutcome {
984 Checked,
986 Skipped,
989 Created,
991 Imported,
994 Recorded,
997 ImportedUnverified,
1000 Applied,
1003 TagsRolledBack,
1008 RolledBack,
1013 AppliedBackupUnsynced,
1020}
1021
1022impl ProjectOutcome {
1023 pub fn as_str(self) -> &'static str {
1025 match self {
1026 ProjectOutcome::Checked => "checked",
1027 ProjectOutcome::Skipped => "skipped",
1028 ProjectOutcome::Created => "created",
1029 ProjectOutcome::Imported => "imported",
1030 ProjectOutcome::Recorded => "recorded",
1031 ProjectOutcome::ImportedUnverified => "imported_unverified",
1032 ProjectOutcome::Applied => "applied",
1033 ProjectOutcome::TagsRolledBack => "tags_rolled_back",
1034 ProjectOutcome::RolledBack => "rolled_back",
1035 ProjectOutcome::AppliedBackupUnsynced => "applied_backup_unsynced",
1036 }
1037 }
1038}
1039
1040pub const PROJECT_UNVERIFIED_SLUG: &str = "deploy_project_unverified";
1046
1047#[derive(Debug, Clone, Serialize)]
1049pub struct DeployProjectsResult {
1050 pub env: String,
1052 pub gateway: String,
1054 pub project: String,
1056 pub release: String,
1058 pub check: bool,
1060 pub action: ProjectAction,
1062 pub would_change: bool,
1064 pub artifact_sha256: String,
1066 pub live_sha256: Option<String>,
1068 pub members: MemberDelta,
1070 pub removed: Vec<String>,
1072 pub removed_count: usize,
1074 pub excluded: Vec<String>,
1079 pub excluded_count: usize,
1081 pub unmanaged: Vec<String>,
1084 pub unmanaged_count: usize,
1086 pub parent: Option<String>,
1089 pub previous_release: Option<String>,
1092 pub recorded_sha256: Option<String>,
1095 pub live_matches_last_deploy: Option<bool>,
1098 pub allow_downgrade: bool,
1100 pub allow_first_change: Option<String>,
1102 pub warnings: Vec<String>,
1104 pub run_id: Option<String>,
1106 pub run_dir: Option<String>,
1109 pub record_written: bool,
1111 pub restore_command: Option<String>,
1115 pub children: Vec<ChildReport>,
1120 pub missing_tags: Option<Vec<MissingTag>>,
1126 pub missing_tags_count: Option<usize>,
1128 pub missing_tags_unknown_providers: Option<Vec<UnknownProvider>>,
1133 pub missing_tags_unknown_providers_count: Option<usize>,
1135 pub missing_tags_unavailable: Option<Vec<TagsUnavailable>>,
1137 pub missing_tags_limit: &'static str,
1139 pub tags: Option<Box<TagsReport>>,
1143 pub outcome: ProjectOutcome,
1145 pub ok: bool,
1147 pub slug: Option<String>,
1149}
1150
1151#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
1153#[serde(rename_all = "snake_case")]
1154pub enum ChildStatus {
1155 Ok,
1157 Mismatch,
1159 Unverified,
1162}
1163
1164impl ChildStatus {
1165 pub fn as_str(self) -> &'static str {
1167 match self {
1168 ChildStatus::Ok => "ok",
1169 ChildStatus::Mismatch => "mismatch",
1170 ChildStatus::Unverified => "unverified",
1171 }
1172 }
1173}
1174
1175#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
1178pub struct ChildReport {
1179 pub name: String,
1181 pub status: ChildStatus,
1183 pub release: Option<String>,
1185 pub recorded_sha256: Option<String>,
1187 pub live_sha256: Option<String>,
1190}
1191
1192impl DeployProjectsResult {
1193 pub fn failure_exit_code(&self) -> Option<u8> {
1198 matches!(
1199 self.outcome,
1200 ProjectOutcome::ImportedUnverified
1201 | ProjectOutcome::TagsRolledBack
1202 | ProjectOutcome::RolledBack
1203 | ProjectOutcome::AppliedBackupUnsynced
1204 )
1205 .then_some(6)
1206 }
1207}
1208
1209fn shell_word(text: &str) -> String {
1212 let plain = !text.is_empty()
1213 && text.chars().all(|c| {
1214 c.is_ascii_alphanumeric() || matches!(c, '_' | '-' | '.' | '/' | ':' | '@' | '+' | ',')
1215 });
1216 if plain {
1217 text.to_string()
1218 } else {
1219 format!("'{}'", text.replace('\'', "'\\''"))
1220 }
1221}
1222
1223fn restore_command(project: &str, run_dir_rel: &str) -> String {
1228 format!(
1229 "ign project import {} --file {} --collision-policy overwrite --yes",
1230 shell_word(project),
1231 shell_word(&format!("{run_dir_rel}/export/{project}.zip")),
1232 )
1233}
1234
1235fn capped(mut items: Vec<String>) -> (Vec<String>, usize) {
1236 let count = items.len();
1237 items.truncate(LIST_CAP);
1238 (items, count)
1239}
1240
1241fn cap_list<T>(mut items: Vec<T>) -> (Vec<T>, usize) {
1243 let count = items.len();
1244 items.truncate(LIST_CAP);
1245 (items, count)
1246}
1247
1248#[derive(Debug, Clone, Default, PartialEq, Eq)]
1251pub struct TagCheck {
1252 pub missing: Option<Vec<MissingTag>>,
1257 pub unknown_providers: Option<Vec<UnknownProvider>>,
1261 pub unavailable: Vec<TagsUnavailable>,
1264}
1265
1266fn unavailable_reason(what: &str, err: &CoreError) -> String {
1268 format!("{what} ({})", err.code())
1269}
1270
1271const UNRESOLVED_PATHS_SHOWN: usize = 3;
1273
1274fn unresolved_reason(paths: &std::collections::BTreeSet<String>) -> String {
1279 let shown: Vec<&str> = paths
1280 .iter()
1281 .take(UNRESOLVED_PATHS_SHOWN)
1282 .map(String::as_str)
1283 .collect();
1284 let rest = paths.len() - shown.len();
1285 let mut reason = format!(
1286 "the provider's tag export does not resolve the UDT type of {} tag path(s): {}",
1287 paths.len(),
1288 shown.join(", ")
1289 );
1290 if rest > 0 {
1291 reason.push_str(&format!(" and {rest} more"));
1292 }
1293 reason
1294}
1295
1296fn strip_property(path: &str) -> Option<&str> {
1299 let start = path.rfind('/').map_or(0, |at| at + 1);
1300 let dot = path[start..].find('.')?;
1301 (dot > 0).then(|| &path[..start + dot])
1302}
1303
1304fn provider_shaped(name: &str) -> bool {
1309 name.chars().any(char::is_alphabetic)
1310 && name
1311 .chars()
1312 .all(|c| c.is_alphanumeric() || matches!(c, '_' | '-' | ' '))
1313 && !name.starts_with(' ')
1314 && !name.ends_with(' ')
1315 && !name.eq_ignore_ascii_case("client")
1316}
1317
1318struct UnknownAcc {
1320 provider: String,
1321 paths: std::collections::BTreeSet<String>,
1322 files: std::collections::BTreeSet<String>,
1323}
1324
1325struct MissingAcc {
1327 path: String,
1328 files: std::collections::BTreeSet<String>,
1329}
1330
1331pub async fn check_tag_references(
1344 zip: &[u8],
1345 excluded: &[String],
1346 gateway: &dyn ProjectGateway,
1347) -> TagCheck {
1348 check_tag_references_with(zip, excluded, gateway, &TagReads::default()).await
1349}
1350
1351#[derive(Debug, Default)]
1354pub struct TagReads {
1355 pub names: Option<Vec<String>>,
1357 pub exports: std::collections::BTreeMap<String, Vec<u8>>,
1359}
1360
1361pub async fn check_tag_references_with(
1364 zip: &[u8],
1365 excluded: &[String],
1366 gateway: &dyn ProjectGateway,
1367 reads: &TagReads,
1368) -> TagCheck {
1369 let refs = match scan_artifact(zip) {
1370 Ok(refs) => refs,
1371 Err(err) => {
1372 return TagCheck {
1373 missing: None,
1374 unknown_providers: None,
1375 unavailable: vec![TagsUnavailable {
1376 provider: None,
1377 reason: unavailable_reason("the project files could not be scanned", &err),
1378 }],
1379 };
1380 }
1381 };
1382 let mut check = check_refs(&refs, excluded, gateway, reads).await;
1383 if let Some(reason) = refs.cut_reason() {
1384 check.unavailable.insert(
1385 0,
1386 TagsUnavailable {
1387 provider: None,
1388 reason,
1389 },
1390 );
1391 check.missing = check.missing.filter(|list| !list.is_empty());
1392 check.unknown_providers = check.unknown_providers.filter(|list| !list.is_empty());
1393 }
1394 check
1395}
1396
1397async fn check_refs(
1399 refs: &TagRefs,
1400 excluded: &[String],
1401 gateway: &dyn ProjectGateway,
1402 reads: &TagReads,
1403) -> TagCheck {
1404 let mut by_provider: std::collections::BTreeMap<String, Vec<&TagRef>> =
1408 std::collections::BTreeMap::new();
1409 for tag in refs.iter() {
1410 let fold = tag.provider.to_ascii_lowercase();
1411 if exclude::is_excluded(&fold, excluded) {
1412 continue;
1413 }
1414 by_provider.entry(fold).or_default().push(tag);
1415 }
1416 if by_provider.is_empty() {
1417 return TagCheck {
1418 missing: Some(Vec::new()),
1419 unknown_providers: Some(Vec::new()),
1420 unavailable: Vec::new(),
1421 };
1422 }
1423 let listed_names = match &reads.names {
1424 Some(names) => Ok(names.clone()),
1425 None => gateway.tag_provider_names().await,
1426 };
1427 let names = match listed_names {
1428 Ok(names) => names,
1429 Err(err) => {
1430 return TagCheck {
1431 missing: None,
1432 unknown_providers: None,
1433 unavailable: vec![TagsUnavailable {
1434 provider: None,
1435 reason: unavailable_reason(
1436 "the gateway's tag providers could not be listed",
1437 &err,
1438 ),
1439 }],
1440 };
1441 }
1442 };
1443 let mut listed: std::collections::BTreeMap<String, String> = std::collections::BTreeMap::new();
1444 for name in names {
1445 let fold = name.to_ascii_lowercase();
1446 match listed.get(&fold) {
1447 Some(kept) if kept.as_bytes() <= name.as_bytes() => {}
1448 _ => {
1449 listed.insert(fold, name);
1450 }
1451 }
1452 }
1453
1454 let mut found: std::collections::BTreeMap<(String, String), MissingAcc> =
1455 std::collections::BTreeMap::new();
1456 let mut unknown: std::collections::BTreeMap<String, UnknownAcc> =
1457 std::collections::BTreeMap::new();
1458 let mut unavailable = Vec::new();
1459 let (mut needed, mut checked) = (0usize, 0usize);
1461 for (fold, tags) in &by_provider {
1462 let Some(provider) = listed.get(fold) else {
1463 if provider_shaped(fold) {
1466 let acc = unknown.entry(fold.clone()).or_insert_with(|| UnknownAcc {
1467 provider: tags[0].provider.clone(),
1468 paths: std::collections::BTreeSet::new(),
1469 files: std::collections::BTreeSet::new(),
1470 });
1471 for tag in tags {
1472 if tag.provider < acc.provider {
1473 acc.provider = tag.provider.clone();
1474 }
1475 let path = strip_property(&tag.path).unwrap_or(&tag.path);
1476 acc.paths.insert(path.to_lowercase());
1477 acc.files.extend(tag.files.iter().cloned());
1478 while acc.files.len() > FILES_PER_ENTRY {
1479 acc.files.pop_last();
1480 }
1481 }
1482 }
1483 continue;
1484 };
1485 needed += 1;
1486 let read = match reads.exports.get(provider) {
1487 Some(bytes) => Ok(bytes.clone()),
1488 None => gateway.tag_export(provider).await,
1489 };
1490 let bytes = match read {
1491 Ok(bytes) => bytes,
1492 Err(err) => {
1493 unavailable.push(TagsUnavailable {
1494 provider: Some(provider.clone()),
1495 reason: unavailable_reason("the provider's tag export could not be read", &err),
1496 });
1497 continue;
1498 }
1499 };
1500 let tree = match TagTree::parse(&bytes, provider) {
1501 Ok(tree) => tree,
1502 Err(err) => {
1503 unavailable.push(TagsUnavailable {
1504 provider: Some(provider.clone()),
1505 reason: unavailable_reason("the provider's tag export is not a tag tree", &err),
1506 });
1507 continue;
1508 }
1509 };
1510 drop(bytes);
1511 let mut unresolved: std::collections::BTreeSet<String> = std::collections::BTreeSet::new();
1514 for tag in tags {
1515 let full = tree.lookup(&tag.path);
1516 let answer = match (full, strip_property(&tag.path)) {
1517 (Lookup::Found, _) => Lookup::Found,
1518 (other, None) => other,
1519 (full, Some(stripped)) => match (full, tree.lookup(stripped)) {
1520 (_, Lookup::Found) => Lookup::Found,
1521 (Lookup::Missing, Lookup::Missing) => Lookup::Missing,
1522 _ => Lookup::Unknown,
1523 },
1524 };
1525 let reported = match answer {
1526 Lookup::Found => {
1527 checked += 1;
1528 continue;
1529 }
1530 Lookup::Unknown => {
1531 unresolved.insert(format!("[{}]{}", tag.provider, tag.path));
1532 continue;
1533 }
1534 Lookup::Missing => {
1535 checked += 1;
1536 strip_property(&tag.path).unwrap_or(&tag.path)
1537 }
1538 };
1539 let display = format!("[{}]{}", tag.provider, reported);
1540 let acc = found
1541 .entry((fold.clone(), reported.to_lowercase()))
1542 .or_insert_with(|| MissingAcc {
1543 path: display.clone(),
1544 files: std::collections::BTreeSet::new(),
1545 });
1546 if display < acc.path {
1547 acc.path = display;
1548 }
1549 acc.files.extend(tag.files.iter().cloned());
1550 while acc.files.len() > FILES_PER_ENTRY {
1551 acc.files.pop_last();
1552 }
1553 }
1554 if !unresolved.is_empty() {
1555 unavailable.push(TagsUnavailable {
1556 provider: Some(provider.clone()),
1557 reason: unresolved_reason(&unresolved),
1558 });
1559 }
1560 }
1561 let missing = if needed > 0 && checked == 0 {
1562 None
1563 } else {
1564 let mut list: Vec<MissingTag> = found
1565 .into_values()
1566 .map(|acc| MissingTag {
1567 path: acc.path,
1568 files: acc.files.into_iter().collect(),
1569 })
1570 .collect();
1571 list.sort_by(|a, b| a.path.as_bytes().cmp(b.path.as_bytes()));
1572 Some(list)
1573 };
1574 let mut unknown_providers: Vec<UnknownProvider> = unknown
1575 .into_values()
1576 .map(|acc| UnknownProvider {
1577 provider: acc.provider,
1578 path_count: acc.paths.len(),
1579 files: acc.files.into_iter().collect(),
1580 })
1581 .collect();
1582 unknown_providers.sort_by(|a, b| a.provider.as_bytes().cmp(b.provider.as_bytes()));
1583 TagCheck {
1584 missing,
1585 unknown_providers: Some(unknown_providers),
1586 unavailable,
1587 }
1588}
1589
1590const WARNING_PROVIDERS_SHOWN: usize = 5;
1592
1593fn tag_warnings(project: &str, gateway: &str, check: &TagCheck) -> Vec<String> {
1596 let mut warnings = Vec::new();
1597 let missing = check.missing.as_ref().map_or(0, Vec::len);
1598 if missing > 0 {
1599 warnings.push(format!(
1600 "project {project}: {missing} tag path(s) the project uses are not on gateway {gateway}; \
1601 ign deploy projects does not deploy tags, and paths built at runtime aren't checked"
1602 ));
1603 }
1604 if let Some(unknown) = check.unknown_providers.as_ref().filter(|u| !u.is_empty()) {
1605 let shown: Vec<&str> = unknown
1606 .iter()
1607 .take(WARNING_PROVIDERS_SHOWN)
1608 .map(|u| u.provider.as_str())
1609 .collect();
1610 let rest = unknown.len().saturating_sub(WARNING_PROVIDERS_SHOWN);
1611 warnings.push(format!(
1612 "project {project}: tag paths the project uses name {} provider(s) that gateway {gateway} does not have: {}{}; \
1613 ign deploy projects does not deploy tags, and paths built at runtime aren't checked",
1614 unknown.len(),
1615 shown.join(", "),
1616 if rest > 0 {
1617 format!(" and {rest} more")
1618 } else {
1619 String::new()
1620 }
1621 ));
1622 }
1623 for entry in &check.unavailable {
1624 warnings.push(format!(
1625 "project {project}: tag paths were not checked for {}: {}",
1626 entry.provider.as_deref().unwrap_or("every provider"),
1627 entry.reason
1628 ));
1629 }
1630 warnings
1631}
1632
1633fn parent_link(record: &ProjectRecord) -> ParentLink {
1635 ParentLink {
1636 name: record.name.clone(),
1637 parent: record.parent.clone().filter(|p| !p.is_empty()),
1638 inheritable: record.inheritable,
1639 invalid_parent: record.invalid_parent,
1640 }
1641}
1642
1643fn live_digest(project: &str, zip: &[u8]) -> Result<CanonicalDigest, CoreError> {
1646 canonical_digest(zip).map_err(|err| match err {
1647 CoreError::DeployTreeInvalid(_) => CoreError::Internal(format!(
1648 "the gateway's export of project {project} holds no members"
1649 )),
1650 other => other,
1651 })
1652}
1653
1654#[derive(Debug, Clone)]
1656pub enum ProjectsPrepared {
1657 Ready {
1660 prose: String,
1663 plan: DeployProjectsResult,
1666 tag_records: std::collections::BTreeMap<String, TagRecord>,
1671 },
1672 Unchanged(DeployProjectsResult),
1675}
1676
1677impl ProjectsPrepared {
1678 pub fn into_check_result(self) -> DeployProjectsResult {
1681 let (ProjectsPrepared::Ready {
1682 plan: mut result, ..
1683 }
1684 | ProjectsPrepared::Unchanged(mut result)) = self;
1685 result.check = true;
1686 result.outcome = ProjectOutcome::Checked;
1687 result
1688 }
1689}
1690
1691fn plan_prose(plan: &DeployProjectsResult) -> String {
1696 let action = match plan.action {
1697 ProjectAction::Create => "create project",
1698 ProjectAction::Import => "overwrite-import project",
1699 ProjectAction::Record => "record the release only for project",
1700 ProjectAction::Skip => "leave project",
1701 };
1702 let mut lines = vec![format!(
1703 "deploy projects will {action} {} on gateway {} (env {}, release {})",
1704 plan.project, plan.gateway, plan.env, plan.release
1705 )];
1706 if plan.action == ProjectAction::Record {
1707 lines.push("the project itself is not changed".to_string());
1708 } else {
1709 lines.push(format!(
1710 "members: {} added, {} changed, {} removed, {} same",
1711 plan.members.added, plan.members.changed, plan.members.removed, plan.members.same
1712 ));
1713 }
1714 if plan.removed_count > 0 {
1715 lines.push(format!(
1716 "the import deletes {} member(s) the repo does not have:",
1717 plan.removed_count
1718 ));
1719 for path in plan.removed.iter().take(PROSE_REMOVED_SHOWN) {
1720 lines.push(format!(" {path}"));
1721 }
1722 if plan.removed_count > PROSE_REMOVED_SHOWN {
1723 lines.push(format!(
1724 " ... and {} more",
1725 plan.removed_count - PROSE_REMOVED_SHOWN
1726 ));
1727 }
1728 }
1729 let missing = plan.missing_tags_count.unwrap_or(0);
1732 if missing > 0 {
1733 lines.push(format!(
1734 "{missing} tag path(s) the project uses are not on gateway {} (this verb does not deploy tags):",
1735 plan.gateway
1736 ));
1737 for tag in plan.missing_tags.iter().flatten().take(PROSE_REMOVED_SHOWN) {
1738 lines.push(format!(" {}", tag.path));
1739 }
1740 if missing > PROSE_REMOVED_SHOWN {
1741 lines.push(format!(" ... and {} more", missing - PROSE_REMOVED_SHOWN));
1742 }
1743 }
1744 let unknown = plan.missing_tags_unknown_providers_count.unwrap_or(0);
1745 if unknown > 0 {
1746 lines.push(format!(
1747 "{unknown} provider(s) the project uses are not on gateway {}:",
1748 plan.gateway
1749 ));
1750 for entry in plan
1751 .missing_tags_unknown_providers
1752 .iter()
1753 .flatten()
1754 .take(PROSE_REMOVED_SHOWN)
1755 {
1756 lines.push(format!(" {}", entry.provider));
1757 }
1758 if unknown > PROSE_REMOVED_SHOWN {
1759 lines.push(format!(" ... and {} more", unknown - PROSE_REMOVED_SHOWN));
1760 }
1761 }
1762 for entry in plan.missing_tags_unavailable.iter().flatten() {
1763 lines.push(format!(
1764 "tags NOT CHECKED for {}: {}",
1765 entry.provider.as_deref().unwrap_or("every provider"),
1766 entry.reason
1767 ));
1768 }
1769 if let Some(tags) = &plan.tags {
1770 lines.push(format!(
1771 "owned tag providers ({}), deployed first, then the project, in one run:",
1772 tags.providers.len()
1773 ));
1774 for p in &tags.providers {
1775 lines.push(format!(
1776 " {}: {} added, {} changed, {} same, {} live-only (kept)",
1777 p.provider, p.added, p.changed, p.same, p.drift
1778 ));
1779 }
1780 for p in &tags.providers {
1781 let extra = [
1782 (p.kept_values, "kept values (live value kept)"),
1783 (p.removed, "removed"),
1784 (p.already_absent, "already absent"),
1785 (p.recreated, "recreated"),
1786 (p.non_canonical, "non-canonical file finding(s)"),
1787 ];
1788 let said: Vec<String> = extra
1789 .iter()
1790 .filter(|(n, _)| *n > 0)
1791 .map(|(n, what)| format!("{n} {what}"))
1792 .collect();
1793 if !said.is_empty() {
1794 lines.push(format!(" {}: {}", p.provider, said.join(", ")));
1795 }
1796 if p.first_change {
1797 lines.push(format!(
1798 " {}: first deploy of the provider by this tool (--allow-first-change)",
1799 p.provider
1800 ));
1801 } else if p.record_only {
1802 lines.push(format!(
1803 " {}: first deploy of the provider by this tool, every node equal",
1804 p.provider
1805 ));
1806 }
1807 }
1808 let list = |lines: &mut Vec<String>, verb: &str, paths: &[String], count: usize| {
1809 for path in paths.iter().take(PROSE_REMOVED_SHOWN) {
1810 lines.push(format!(" {verb} {path}"));
1811 }
1812 if count > PROSE_REMOVED_SHOWN {
1813 lines.push(format!(
1814 " ... and {} more {verb}",
1815 count - PROSE_REMOVED_SHOWN
1816 ));
1817 }
1818 };
1819 list(&mut lines, "add", &tags.added, tags.added_count);
1820 let changed: Vec<String> = tags
1821 .changed
1822 .iter()
1823 .map(|c| {
1824 format!(
1825 "{} ({}){}",
1826 c.path,
1827 c.properties.join(", "),
1828 if c.restart_likely {
1829 ", restart likely"
1830 } else {
1831 ""
1832 }
1833 )
1834 })
1835 .collect();
1836 list(&mut lines, "change", &changed, tags.changed_count);
1837 list(
1840 &mut lines,
1841 "remove (with --allow-destructive)",
1842 &tags.removed,
1843 tags.removed_count,
1844 );
1845 let groups: Vec<String> = tags
1847 .groups
1848 .iter()
1849 .flat_map(|g| g.removed.iter().map(|n| full_tag_path(&g.provider, n)))
1850 .collect();
1851 list(
1852 &mut lines,
1853 "remove tag group (with --allow-destructive)",
1854 &groups,
1855 groups.len(),
1856 );
1857 list(
1858 &mut lines,
1859 "kept value",
1860 &tags.kept_values,
1861 tags.kept_values_count,
1862 );
1863 for warning in &tags.warnings {
1864 lines.push(format!(" warning: {warning}"));
1865 }
1866 lines.push(format!("tags_sha256 {}", tags.tags_sha256));
1867 lines.push(format!("tags plan_sha256 {}", tags.plan_sha256));
1868 }
1869 if plan.action == ProjectAction::Import {
1870 lines.push("the live project is saved into a run directory before the import".to_string());
1871 }
1872 lines.push(format!("artifact_sha256 {}", plan.artifact_sha256));
1873 lines.join("\n")
1874}
1875
1876const PROSE_REMOVED_SHOWN: usize = 20;
1878
1879pub async fn deploy_projects_prepare(
1888 loaded: &LoadedProject,
1889 input: &ProjectsInput,
1890 surfaces: &ProjectsSurfaces<'_>,
1891) -> Result<ProjectsPrepared, CoreError> {
1892 let project = &loaded.file.project;
1893 let gateway = surfaces.gateway;
1894 let gateway_name = &loaded.target.gateway;
1895
1896 let git = surfaces.host.git_entry(project).await?;
1897 if git.is_present() {
1898 return Err(CoreError::ProjectGitManaged {
1899 project: project.clone(),
1900 location: git_entry_location(project),
1901 kind: git.as_str().to_string(),
1902 });
1903 }
1904
1905 let record = match surfaces
1909 .host
1910 .read_release_record(gateway_name, project)
1911 .await?
1912 {
1913 None => None,
1914 Some(bytes) => {
1915 let location = release_record_location(
1916 LedgerRoot::for_node(&loaded.target.nodes.master),
1917 gateway_name,
1918 project,
1919 );
1920 Some(ReleaseRecord::parse(
1921 &bytes,
1922 &location,
1923 gateway_name,
1924 project,
1925 )?)
1926 }
1927 };
1928 if let Some(record) = &record
1929 && loaded.release.order_against(&record.release()?) == ReleaseOrder::Older
1930 && !input.allow_downgrade
1931 {
1932 return Err(CoreError::DestructiveChangeRefused {
1933 gate: "downgrade".to_string(),
1934 keys: vec![project.clone()],
1935 count: 1,
1936 });
1937 }
1938
1939 let tag_records = read_tag_records(loaded, surfaces.host).await?;
1943 let mut warnings = Vec::new();
1944 let journal = read_project_journal(loaded, surfaces.host).await?;
1947 if let Some(journal) = &journal {
1948 warnings.push(format!(
1949 "project {project}: run {run} left its host journal ({state}) at {location}; an apply \
1950 refuses until `{command}` undoes it",
1951 run = journal.run_id,
1952 state = journal.state.as_str(),
1953 location = project_journal_location(
1954 LedgerRoot::for_node(&loaded.target.nodes.master),
1955 gateway_name,
1956 project
1957 ),
1958 command = rollback_run_command(&input.env, &journal.run_id),
1959 ));
1960 }
1961
1962 let records = gateway.projects().await?;
1963 if let Some(parent) = &loaded.meta.parent {
1967 let all: Vec<ParentLink> = records.iter().map(parent_link).collect();
1968 let (chain, bound_hit) = parent_chain(parent, &all);
1969 check_parent(project, parent, &chain, bound_hit).map_err(|reason| {
1970 CoreError::ProjectParentInvalid {
1971 project: project.clone(),
1972 parent: parent.clone(),
1973 reason: reason.as_str().to_string(),
1974 }
1975 })?;
1976 if chain
1980 .first()
1981 .is_some_and(|p| p.invalid_parent == Some(true))
1982 {
1983 warnings.push(format!(
1984 "project {project}: the gateway reports that parent {parent} has an invalid parent of its own"
1985 ));
1986 }
1987 }
1988 let repo_parent = loaded.meta.parent.as_deref().filter(|p| !p.is_empty());
1992 if let Some(live_record) = records.iter().find(|r| r.name == *project) {
1993 let live_parent = live_record.parent.as_deref().filter(|p| !p.is_empty());
1994 if live_parent != repo_parent {
1995 warnings.push(format!(
1996 "project {project}: importing this project changes its parent from {} to {}",
1997 live_parent.unwrap_or("none"),
1998 repo_parent.unwrap_or("none")
1999 ));
2000 }
2001 }
2002 let mut others: Vec<String> = records
2003 .into_iter()
2004 .map(|r| r.name)
2005 .filter(|name| name != project)
2006 .collect();
2007 others.sort_by(|a, b| a.as_bytes().cmp(b.as_bytes()));
2008 others.dedup();
2009
2010 let live = match gateway.project_find(project).await? {
2011 None => None,
2012 Some(_) => {
2013 let zip = gateway.export_bytes(project).await?;
2014 Some(live_digest(project, &zip)?)
2015 }
2016 };
2017
2018 let artifact = &loaded.artifact.digest;
2019 let equal = live.as_ref().is_some_and(|live| {
2020 !live.members.is_empty() && !artifact.members.is_empty() && live.sha256 == artifact.sha256
2021 });
2022 let action = match &live {
2023 None => ProjectAction::Create,
2024 Some(_) if equal => match &record {
2027 Some(record) if record.release()?.version == loaded.release.version => {
2028 ProjectAction::Skip
2029 }
2030 _ => ProjectAction::Record,
2031 },
2032 Some(_) => ProjectAction::Import,
2033 };
2034 if action == ProjectAction::Import && record.is_none() && input.allow_first_change.is_none() {
2037 return Err(CoreError::DestructiveChangeRefused {
2038 gate: "first_deploy".to_string(),
2039 keys: vec![project.clone()],
2040 count: 1,
2041 });
2042 }
2043 let (tag_plan, tag_reads, provider_settings, tag_live) =
2047 match plan_owned_tags(loaded, gateway, &tag_records).await? {
2048 Some(owned) => (Some(owned.plan), owned.reads, owned.settings, owned.live),
2049 None => (None, TagReads::default(), Vec::new(), ProviderReads::new()),
2050 };
2051 let tag_secret_warnings = match &tag_plan {
2054 Some(plan) => tag_gates(loaded, input, plan)?,
2055 None => Vec::new(),
2056 };
2057 let groups = match &tag_plan {
2061 Some(plan) => plan_owned_groups(loaded, surfaces, plan, &tag_records, &tag_live).await?,
2062 None => Vec::new(),
2063 };
2064 drop(tag_live);
2065 let tag_check = check_tag_references_with(
2069 &loaded.artifact.zip,
2070 &loaded.tag_exclusions,
2071 gateway,
2072 &tag_reads,
2073 )
2074 .await;
2075 drop(tag_reads);
2076 let live_matches_last_deploy = match (&record, &live) {
2077 (Some(record), Some(live)) => Some(record.artifact_sha256 == live.sha256),
2078 _ => None,
2079 };
2080 if let (Some(record), Some(false)) = (&record, live_matches_last_deploy) {
2081 warnings.push(format!(
2082 "project {project}: the live project no longer matches the last release this tool deployed ({})",
2083 record.release
2084 ));
2085 }
2086 warnings.extend(tag_warnings(project, gateway_name, &tag_check));
2087 let (members, removed) = member_delta(artifact, live.as_ref());
2088 let (removed, removed_count) = capped(removed);
2089 let (excluded, excluded_count) = capped(loaded.artifact.excluded.clone());
2090 let (unmanaged, unmanaged_count) = capped(others);
2091 let (missing_tags, missing_tags_count) = match tag_check.missing {
2092 Some(list) => {
2093 let (list, count) = cap_list(list);
2094 (Some(list), Some(count))
2095 }
2096 None => (None, None),
2097 };
2098 let (missing_tags_unknown_providers, missing_tags_unknown_providers_count) =
2099 match tag_check.unknown_providers {
2100 Some(list) => {
2101 let (list, count) = cap_list(list);
2102 (Some(list), Some(count))
2103 }
2104 None => (None, None),
2105 };
2106 let missing_tags_unavailable =
2107 (!tag_check.unavailable.is_empty()).then_some(tag_check.unavailable);
2108 let settings_change = provider_settings
2109 .iter()
2110 .any(|p| p.action != ProviderSettingsAction::Same);
2111 let would_change = action.would_change()
2112 || tag_plan.as_ref().is_some_and(TagPlan::would_change)
2113 || settings_change
2114 || groups.iter().any(GroupPlan::would_change);
2115 let delete_route = match (&tag_plan, &surfaces.tag_delete) {
2118 (None, _) => None,
2119 (Some(_), TagDeleteSurface::Configured(route)) => Some(RouteState::of(
2120 &surfaces.tag_delete,
2121 Some(route.handshake().await),
2122 )),
2123 (Some(_), surface) => Some(RouteState::of(surface, None)),
2124 }
2125 .map(|state| state.report(&owned_names(loaded)));
2126 let tags = tag_plan.map(|plan| {
2127 let mut report = TagsReport::from_plan(&plan, &loaded.non_canonical, LIST_CAP);
2128 report.delete_route = delete_route;
2129 report.warnings.clone_from(&loaded.ownership_warnings);
2130 report.warnings.extend(tag_secret_warnings);
2131 report.provider_settings = provider_settings;
2132 report.groups = groups;
2133 report.journal = journal.as_ref().map(|j| TagJournal {
2134 run_id: j.run_id.clone(),
2135 state: j.state.as_str().to_string(),
2136 started_at: j.started_at.clone(),
2137 });
2138 Box::new(report)
2139 });
2140 let outcome = if !would_change {
2141 ProjectOutcome::Skipped
2142 } else if tags.is_some() {
2143 ProjectOutcome::Applied
2144 } else {
2145 match action {
2146 ProjectAction::Create => ProjectOutcome::Created,
2147 ProjectAction::Skip => ProjectOutcome::Skipped,
2148 ProjectAction::Import => ProjectOutcome::Imported,
2149 ProjectAction::Record => ProjectOutcome::Recorded,
2150 }
2151 };
2152
2153 let plan = DeployProjectsResult {
2154 env: input.env.clone(),
2155 gateway: loaded.target.gateway.clone(),
2156 project: project.clone(),
2157 release: input.release.clone(),
2158 check: false,
2159 action,
2160 would_change,
2161 artifact_sha256: artifact.sha256.clone(),
2162 live_sha256: live.map(|l| l.sha256),
2163 members,
2164 removed,
2165 removed_count,
2166 excluded,
2167 excluded_count,
2168 unmanaged,
2169 unmanaged_count,
2170 parent: loaded.meta.parent.clone(),
2171 previous_release: record.as_ref().map(|r| r.release.clone()),
2172 recorded_sha256: record.map(|r| r.artifact_sha256),
2173 live_matches_last_deploy,
2174 allow_downgrade: input.allow_downgrade,
2175 allow_first_change: input.allow_first_change.clone(),
2176 warnings,
2177 run_id: None,
2178 run_dir: None,
2179 record_written: false,
2180 restore_command: None,
2181 children: Vec::new(),
2182 missing_tags,
2183 missing_tags_count,
2184 missing_tags_unknown_providers,
2185 missing_tags_unknown_providers_count,
2186 missing_tags_unavailable,
2187 missing_tags_limit: MISSING_TAGS_LIMIT,
2188 tags,
2189 outcome,
2190 ok: true,
2191 slug: None,
2192 };
2193 Ok(if !would_change {
2194 ProjectsPrepared::Unchanged(plan)
2195 } else {
2196 ProjectsPrepared::Ready {
2197 prose: plan_prose(&plan),
2198 plan,
2199 tag_records,
2200 }
2201 })
2202}
2203
2204struct OwnedTags {
2206 plan: TagPlan,
2207 reads: TagReads,
2208 live: ProviderReads,
2211 settings: Vec<ProviderSettingsPlan>,
2213}
2214
2215async fn plan_owned_tags(
2224 loaded: &LoadedProject,
2225 gateway: &dyn ProjectGateway,
2226 records: &std::collections::BTreeMap<String, TagRecord>,
2227) -> Result<Option<OwnedTags>, CoreError> {
2228 if loaded.owned_tags.is_empty() {
2229 return Ok(None);
2230 }
2231 let names = gateway.tag_provider_names().await?;
2232 let mut findings = Vec::new();
2233 for (i, provider) in loaded.owned_tags.iter().enumerate() {
2234 if names.contains(&provider.name)
2235 || !names.iter().any(|n| n.eq_ignore_ascii_case(&provider.name))
2236 {
2237 continue;
2238 }
2239 findings.push(crate::error::DeployFinding {
2240 file: PROJECT_FILE_REL.to_string(),
2241 line: None,
2242 column: None,
2243 message: format!(
2244 "tag_providers[{i}]: gateway {} has no tag provider named {} (it lists one whose \
2245 name differs only in letter case; the names must match exactly)",
2246 loaded.target.gateway, provider.name
2247 ),
2248 });
2249 }
2250 if !findings.is_empty() {
2251 return Err(CoreError::deploy_findings(findings));
2252 }
2253 let mut settings = Vec::new();
2254 for provider in sorted(&loaded.owned_tags) {
2255 let listed = names.contains(&provider.name);
2256 let record = if listed {
2257 let record = gateway.tag_provider_record(&provider.name).await?;
2258 if record.is_none() {
2259 return Err(CoreError::Internal(format!(
2260 "the gateway lists tag provider {} but has no record of it",
2261 provider.name
2262 )));
2263 }
2264 record
2265 } else {
2266 None
2267 };
2268 settings.push(plan_provider_settings(
2269 &provider.name,
2270 provider_settings_of(loaded, &provider.name)?,
2271 record.as_ref(),
2272 ));
2273 }
2274 let mut reads = TagReads {
2275 names: Some(names),
2276 exports: std::collections::BTreeMap::new(),
2277 };
2278 let mut live = std::collections::BTreeMap::new();
2279 for provider in &loaded.owned_tags {
2280 if !reads
2281 .names
2282 .as_ref()
2283 .is_some_and(|names| names.contains(&provider.name))
2284 {
2285 continue;
2286 }
2287 let bytes = gateway.tag_export(&provider.name).await?;
2288 live.insert(
2289 provider.name.clone(),
2290 flatten_export(&bytes, &provider.name)?,
2291 );
2292 reads.exports.insert(provider.name.clone(), bytes);
2293 }
2294 let plan = plan_tags(&loaded.owned_tags, &live, records);
2295 let empty = std::collections::BTreeMap::new();
2299 let mut in_use = Vec::new();
2300 for (provider_plan, provider) in plan.providers.iter().zip(sorted(&loaded.owned_tags)) {
2301 let nodes = live.get(&provider.name).unwrap_or(&empty);
2302 for (type_path, referrer) in removed_types_in_use(provider_plan, provider, nodes) {
2303 in_use.push(crate::error::DeployFinding {
2304 file: provider.nodes.get(&referrer).map_or_else(
2305 || full_tag_path(&provider.name, &referrer),
2306 |node| node.file.clone(),
2307 ),
2308 line: None,
2309 column: None,
2310 message: format!(
2311 "the repo removes the type {} and {} still uses it (typeId), so deleting it \
2312 would break a tag that stays; keep the type in the repo or remove its users \
2313 too; nothing was written",
2314 full_tag_path(&provider.name, &type_path),
2315 full_tag_path(&provider.name, &referrer)
2316 ),
2317 });
2318 }
2319 }
2320 if !in_use.is_empty() {
2321 return Err(CoreError::deploy_findings(in_use));
2322 }
2323 Ok(Some(OwnedTags {
2324 plan,
2325 reads,
2326 settings,
2327 live,
2328 }))
2329}
2330
2331fn need_tree<'a>(
2333 surfaces: &ProjectsSurfaces<'a>,
2334 what: &str,
2335) -> Result<&'a dyn TreeReader, CoreError> {
2336 surfaces.tree.ok_or_else(|| {
2337 CoreError::Internal(format!(
2338 "{what} needs the master's host reader, and none was supplied"
2339 ))
2340 })
2341}
2342
2343const UNITS_SHOWN: usize = 10;
2345
2346async fn refuse_mode_tag_units(
2352 loaded: &LoadedProject,
2353 reader: &dyn TreeReader,
2354) -> Result<(), CoreError> {
2355 let mode = &loaded.target.mode;
2356 let dirs: Vec<(String, String)> = crate::deploy::layout::PAIR_UNSYNCED_TAG_TYPES
2357 .iter()
2358 .map(|ty| ("ignition".to_string(), (*ty).to_string()))
2359 .collect();
2360 let listed = reader.list_tree(mode, &dirs).await?;
2361 let mut found: Vec<String> = Vec::new();
2362 for listing in &listed {
2363 for provider in &loaded.file.tag_providers {
2364 let root = format!("{}/{}/{provider}", listing.module, listing.ty);
2365 let below = format!("{root}/");
2366 found.extend(
2367 listing
2368 .entries
2369 .iter()
2370 .map(|e| e.path.as_str())
2371 .filter(|p| *p == root || p.starts_with(&below))
2372 .map(str::to_string),
2373 );
2374 }
2375 }
2376 if found.is_empty() {
2377 return Ok(());
2378 }
2379 found.sort_by(|a, b| a.as_bytes().cmp(b.as_bytes()));
2380 let count = found.len();
2381 let shown: Vec<&str> = found.iter().take(UNITS_SHOWN).map(String::as_str).collect();
2382 Err(CoreError::deploy_tree_invalid(
2383 &loaded.target_file,
2384 None,
2385 None,
2386 format!(
2387 "nodes.master's {mode} collection holds {count} tag definition path(s) of an owned \
2388 provider: {}{}. On a redundant pair the tag import would be written into the {mode} \
2389 collection (C8), which stops the backup's configuration sync; move these units to \
2390 core with a gateway-config deploy (ign deploy apply) first",
2391 shown.join(", "),
2392 if count > shown.len() {
2393 format!(" and {} more", count - shown.len())
2394 } else {
2395 String::new()
2396 }
2397 ),
2398 ))
2399}
2400
2401struct GroupInputs<'a> {
2403 provider: &'a NativeProvider,
2404 summary: Option<std::collections::BTreeMap<String, GroupSummary>>,
2405 sent: Vec<&'a NativeNode>,
2407 removed: std::collections::BTreeSet<String>,
2409}
2410
2411fn group_inputs<'a>(
2413 loaded: &'a LoadedProject,
2414 plan: &TagPlan,
2415) -> Result<Vec<GroupInputs<'a>>, CoreError> {
2416 let mut out = Vec::new();
2417 for (provider_plan, provider) in plan.providers.iter().zip(sorted(&loaded.owned_tags)) {
2418 let summary = match &provider.groups {
2419 Some(value) => Some(parse_group_value(
2420 &format!("tags/{}/{TAG_GROUPS_FILE}", provider.name),
2421 value,
2422 )?),
2423 None => None,
2424 };
2425 let sending: std::collections::BTreeSet<&str> = provider_plan
2426 .added
2427 .iter()
2428 .map(String::as_str)
2429 .chain(provider_plan.changed.iter().map(|c| c.path.as_str()))
2430 .collect();
2431 let sent = provider
2432 .nodes
2433 .values()
2434 .filter(|n| sending.contains(n.path.as_str()))
2435 .collect();
2436 out.push(GroupInputs {
2437 provider,
2438 summary,
2439 sent,
2440 removed: provider_plan.removed.iter().cloned().collect(),
2441 });
2442 }
2443 Ok(out)
2444}
2445
2446fn group_plans(
2456 inputs: &[GroupInputs<'_>],
2457 observed: &crate::deploy::transport::ObservedFiles,
2458 records: &std::collections::BTreeMap<String, TagRecord>,
2459 live_nodes: &ProviderReads,
2460) -> Result<Vec<GroupPlan>, CoreError> {
2461 let mut plans = Vec::new();
2462 let mut in_use = Vec::new();
2463 let empty = BTreeMap::new();
2464 for input in inputs {
2465 let Some(summary) = &input.summary else {
2466 continue;
2467 };
2468 let name = &input.provider.name;
2469 let live = LiveGroups::from_observed(name, observed);
2470 let recorded = records
2471 .get(name)
2472 .map(|r| r.groups.clone())
2473 .unwrap_or_default();
2474 let referenced = referenced_groups(input.provider.nodes.values());
2475 let plan = plan_groups(name, summary, &live, &recorded, &referenced)?;
2476 if !plan.removed.is_empty() {
2477 let users = live_group_users(live_nodes.get(name).unwrap_or(&empty), &input.removed);
2478 for group in &plan.removed {
2479 if let Some(user) = users.get(group) {
2480 in_use.push(crate::error::DeployFinding {
2481 file: format!("tags/{name}/{TAG_GROUPS_FILE}"),
2482 line: None,
2483 column: None,
2484 message: format!(
2485 "the repo drops tag group `{group}` of provider {name}, which this \
2486 project recorded, and the live tag {} still names it, so deleting \
2487 the group would leave that tag without its group; keep the group in \
2488 the summary or change the tag first; nothing was written",
2489 full_tag_path(name, user)
2490 ),
2491 });
2492 }
2493 }
2494 }
2495 plans.push(plan);
2496 }
2497 if !in_use.is_empty() {
2498 return Err(CoreError::deploy_findings(in_use));
2499 }
2500 Ok(plans)
2501}
2502
2503fn providers_needing_groups(inputs: &[GroupInputs<'_>]) -> Vec<String> {
2506 inputs
2507 .iter()
2508 .filter(|input| {
2509 input.summary.is_some() || !referenced_groups(input.sent.iter().copied()).is_empty()
2510 })
2511 .map(|input| input.provider.name.clone())
2512 .collect()
2513}
2514
2515async fn read_groups(
2518 reader: &dyn TreeReader,
2519 mode: &str,
2520 providers: &[String],
2521) -> Result<crate::deploy::transport::ObservedFiles, CoreError> {
2522 let observed = reader.read_tag_groups(mode, providers).await?;
2523 if let Some(first) = observed.unreadable.first() {
2524 return Err(CoreError::TransportFailed {
2525 operation: "read".to_string(),
2526 node: "nodes.master".to_string(),
2527 detail: format!(
2528 "{} tag-group file(s) cannot be read, the first {}/{}",
2529 observed.unreadable.len(),
2530 first.collection,
2531 first.path
2532 ),
2533 });
2534 }
2535 Ok(observed)
2536}
2537
2538async fn plan_owned_groups(
2543 loaded: &LoadedProject,
2544 surfaces: &ProjectsSurfaces<'_>,
2545 plan: &TagPlan,
2546 records: &std::collections::BTreeMap<String, TagRecord>,
2547 live: &ProviderReads,
2548) -> Result<Vec<GroupPlan>, CoreError> {
2549 if loaded.target.nodes.backup.is_some() {
2552 refuse_mode_tag_units(
2553 loaded,
2554 need_tree(surfaces, "the pair's mode-collection check")?,
2555 )
2556 .await?;
2557 }
2558 let inputs = group_inputs(loaded, plan)?;
2559 let providers = providers_needing_groups(&inputs);
2560 if providers.is_empty() {
2561 return Ok(Vec::new());
2562 }
2563 let reader = need_tree(surfaces, "the tag-group read")?;
2564 let observed = read_groups(reader, &loaded.target.mode, &providers).await?;
2565 let mut findings = Vec::new();
2566 let empty = std::collections::BTreeMap::new();
2567 for input in &inputs {
2568 let live = LiveGroups::from_observed(&input.provider.name, &observed).names();
2569 match check_group_refs(
2570 &input.provider.name,
2571 &input.sent,
2572 input.summary.as_ref().unwrap_or(&empty),
2573 &live,
2574 ) {
2575 Ok(()) => {}
2576 Err(CoreError::DeployTreeInvalid(details)) => findings.extend(details.findings),
2577 Err(other) => return Err(other),
2578 }
2579 }
2580 if !findings.is_empty() {
2581 return Err(CoreError::deploy_findings(findings));
2582 }
2583 group_plans(&inputs, &observed, records, live)
2584}
2585
2586fn provider_settings_of<'a>(
2589 loaded: &'a LoadedProject,
2590 provider: &str,
2591) -> Result<&'a ProviderSettings, CoreError> {
2592 loaded
2593 .file
2594 .tag_providers
2595 .iter()
2596 .position(|name| name == provider)
2597 .and_then(|i| loaded.provider_settings.get(i))
2598 .ok_or_else(|| {
2599 CoreError::Internal(format!(
2600 "no settings were loaded for owned tag provider {provider}"
2601 ))
2602 })
2603}
2604
2605async fn read_tag_records(
2611 loaded: &LoadedProject,
2612 host: &dyn ProjectHost,
2613) -> Result<std::collections::BTreeMap<String, TagRecord>, CoreError> {
2614 let gateway = &loaded.target.gateway;
2615 let mut records = std::collections::BTreeMap::new();
2616 for provider in &loaded.file.tag_providers {
2617 let Some(bytes) = host.read_tag_record(gateway, provider).await? else {
2618 continue;
2619 };
2620 let location = tag_record_location(
2621 LedgerRoot::for_node(&loaded.target.nodes.master),
2622 gateway,
2623 provider,
2624 );
2625 let record = TagRecord::parse(&bytes, &location, gateway, provider)?;
2626 if record.project != loaded.file.project {
2627 return Err(CoreError::TagProviderClaimed {
2628 provider: provider.clone(),
2629 claimed_by: format!("project {} (host record)", record.project),
2630 });
2631 }
2632 records.insert(provider.clone(), record);
2633 }
2634 Ok(records)
2635}
2636
2637async fn read_project_journal(
2641 loaded: &LoadedProject,
2642 host: &dyn ProjectHost,
2643) -> Result<Option<ProjectJournal>, CoreError> {
2644 if loaded.file.tag_providers.is_empty() {
2645 return Ok(None);
2646 }
2647 let gateway = &loaded.target.gateway;
2648 let project = &loaded.file.project;
2649 let Some(bytes) = host.read_journal(gateway, project).await? else {
2650 return Ok(None);
2651 };
2652 let location = project_journal_location(
2653 LedgerRoot::for_node(&loaded.target.nodes.master),
2654 gateway,
2655 project,
2656 );
2657 ProjectJournal::parse(&bytes, &location, gateway, project).map(Some)
2658}
2659
2660fn rollback_run_command(env: &str, run_id: &str) -> String {
2662 format!(
2663 "ign deploy projects --env {} --rollback-run {run_id}",
2664 shell_word(env)
2665 )
2666}
2667
2668fn project_journal_present(loaded: &LoadedProject, env: &str, run_id: &str) -> CoreError {
2671 CoreError::DeployJournalPresent {
2672 run_id: run_id.to_string(),
2673 env: env.to_string(),
2674 location: project_journal_location(
2675 LedgerRoot::for_node(&loaded.target.nodes.master),
2676 &loaded.target.gateway,
2677 &loaded.file.project,
2678 ),
2679 recover_with: Some(rollback_run_command(env, run_id)),
2680 }
2681}
2682
2683fn tag_gates(
2700 loaded: &LoadedProject,
2701 input: &ProjectsInput,
2702 plan: &TagPlan,
2703) -> Result<Vec<String>, CoreError> {
2704 let named = |pick: fn(&crate::deploy::tag_plan::ProviderPlan) -> bool| -> Vec<String> {
2705 plan.providers
2706 .iter()
2707 .filter(|p| pick(p))
2708 .map(|p| format!("tag provider {}", p.provider))
2709 .collect()
2710 };
2711 let emptied = named(|p| p.provider_emptied);
2712 if !emptied.is_empty() {
2713 return Err(CoreError::DestructiveChangeRefused {
2714 gate: "provider_emptied".to_string(),
2715 count: emptied.len(),
2716 keys: emptied,
2717 });
2718 }
2719 let first = named(|p| p.first_change);
2720 if !first.is_empty() && input.allow_first_change.is_none() {
2721 return Err(CoreError::DestructiveChangeRefused {
2722 gate: "first_deploy".to_string(),
2723 count: first.len(),
2724 keys: first,
2725 });
2726 }
2727 let mut refused = Vec::new();
2728 let mut warnings = Vec::new();
2729 for (provider_plan, provider) in plan.providers.iter().zip(sorted(&loaded.owned_tags)) {
2730 let added: std::collections::BTreeSet<&str> =
2731 provider_plan.added.iter().map(String::as_str).collect();
2732 let changed: std::collections::BTreeSet<&str> = provider_plan
2733 .changed
2734 .iter()
2735 .map(|c| c.path.as_str())
2736 .collect();
2737 for (path, node) in &provider.nodes {
2738 let sends = if added.contains(path.as_str()) {
2739 Sends::All
2740 } else if changed.contains(path.as_str()) {
2741 Sends::Config
2742 } else {
2743 Sends::Nothing
2744 };
2745 let scan = scan_tag_node(&node.file, path, &to_node(&node.json), sends);
2746 refused.extend(scan.refused);
2747 warnings.extend(scan.warnings);
2748 }
2749 }
2750 if !refused.is_empty() {
2751 return Err(CoreError::embedded_secret(refused, Vec::new()));
2752 }
2753 Ok(warnings)
2754}
2755
2756#[derive(Debug, Clone, PartialEq, Eq)]
2761pub enum ProjectsEvent {
2762 DesignersPoll {
2765 open: usize,
2767 },
2768 DesignersProceed {
2770 sessions: Vec<DesignerSeen>,
2772 },
2773 DesignersSkipped,
2776 RunStarted {
2778 run_id: String,
2780 action: ProjectAction,
2782 },
2783 PreImageExported {
2785 bytes: usize,
2787 },
2788 TagsPreImageExported {
2791 provider: String,
2793 bytes: usize,
2795 },
2796 TagImportSent {
2798 provider: String,
2800 path: String,
2803 policy: &'static str,
2805 nodes: usize,
2807 carry: bool,
2809 },
2810 TagValuesCarried {
2814 provider: String,
2816 path: String,
2818 nodes: usize,
2820 },
2821 TagProviderSettingsApplied {
2824 provider: String,
2826 action: &'static str,
2828 },
2829 TagGroupsWritten {
2832 providers: Vec<String>,
2834 written: usize,
2836 deleted: usize,
2838 },
2839 TagImportAnswered {
2841 provider: String,
2843 clean: bool,
2845 },
2846 TagsVerified {
2849 provider: String,
2851 equal: bool,
2854 },
2855 ImportSent {
2857 overwrite: bool,
2859 },
2860 ImportAnswered,
2862 Verified {
2864 equal: bool,
2866 },
2867 RecordWritten,
2869 RecordFailed {
2871 code: String,
2873 },
2874 TagRecordWritten {
2876 provider: String,
2878 },
2879 TagRecordFailed {
2881 provider: String,
2883 code: String,
2885 },
2886 ChildrenSettling {
2889 children: Vec<String>,
2891 },
2892 ChildChecked {
2894 name: String,
2896 status: ChildStatus,
2898 },
2899 RollbackStarted {
2902 cause: RollbackCause,
2904 },
2905 ProjectRestored {
2908 equal: bool,
2910 },
2911 TagGroupsRestored {
2914 providers: Vec<String>,
2916 },
2917 TagProviderRestored {
2920 provider: String,
2922 action: &'static str,
2924 },
2925 RollbackVerified {
2928 equal: bool,
2930 },
2931 TagDeleteSent {
2935 provider: String,
2937 paths: Vec<String>,
2939 rollback: bool,
2941 },
2942 TagDeleteAnswered {
2944 provider: String,
2946 clean: bool,
2948 deleted: usize,
2950 },
2951 HostPreImageWritten {
2955 files: usize,
2957 },
2958 JournalWritten {
2960 state: &'static str,
2962 },
2963 JournalCleared,
2966 BackupWaitStarted {
2968 budget_s: u64,
2970 },
2971 BackupResync {
2973 ok: bool,
2975 code: Option<String>,
2977 },
2978 BackupVerdict {
2980 converged: bool,
2982 reason: Option<&'static str>,
2985 },
2986 Finished {
2988 outcome: ProjectOutcome,
2990 },
2991}
2992
2993impl ProjectsEvent {
2994 pub fn name(&self) -> &'static str {
2996 match self {
2997 ProjectsEvent::DesignersPoll { .. } => "designers_poll",
2998 ProjectsEvent::DesignersProceed { .. } => "designers_proceed",
2999 ProjectsEvent::DesignersSkipped => "designers_skipped",
3000 ProjectsEvent::RunStarted { .. } => "run_started",
3001 ProjectsEvent::PreImageExported { .. } => "pre_image_exported",
3002 ProjectsEvent::TagsPreImageExported { .. } => "tags_pre_image_exported",
3003 ProjectsEvent::TagImportSent { .. } => "tag_import_sent",
3004 ProjectsEvent::TagValuesCarried { .. } => "tag_values_carried",
3005 ProjectsEvent::TagProviderSettingsApplied { .. } => "tag_provider_settings_applied",
3006 ProjectsEvent::TagGroupsWritten { .. } => "tag_groups_written",
3007 ProjectsEvent::TagImportAnswered { .. } => "tag_import_answered",
3008 ProjectsEvent::TagsVerified { .. } => "tags_verified",
3009 ProjectsEvent::ImportSent { .. } => "import_sent",
3010 ProjectsEvent::ImportAnswered => "import_answered",
3011 ProjectsEvent::Verified { .. } => "verified",
3012 ProjectsEvent::RecordWritten => "record_written",
3013 ProjectsEvent::RecordFailed { .. } => "record_failed",
3014 ProjectsEvent::TagRecordWritten { .. } => "tag_record_written",
3015 ProjectsEvent::TagRecordFailed { .. } => "tag_record_failed",
3016 ProjectsEvent::ChildrenSettling { .. } => "children_settling",
3017 ProjectsEvent::ChildChecked { .. } => "child_checked",
3018 ProjectsEvent::RollbackStarted { .. } => "rollback_started",
3019 ProjectsEvent::ProjectRestored { .. } => "project_restored",
3020 ProjectsEvent::TagGroupsRestored { .. } => "tag_groups_restored",
3021 ProjectsEvent::TagProviderRestored { .. } => "tag_provider_restored",
3022 ProjectsEvent::RollbackVerified { .. } => "rollback_verified",
3023 ProjectsEvent::TagDeleteSent { .. } => "tag_delete_sent",
3024 ProjectsEvent::TagDeleteAnswered { .. } => "tag_delete_answered",
3025 ProjectsEvent::HostPreImageWritten { .. } => "host_pre_image_written",
3026 ProjectsEvent::JournalWritten { .. } => "journal_written",
3027 ProjectsEvent::JournalCleared => "journal_cleared",
3028 ProjectsEvent::BackupWaitStarted { .. } => "backup_wait_started",
3029 ProjectsEvent::BackupResync { .. } => "backup_resync",
3030 ProjectsEvent::BackupVerdict { .. } => "backup_verdict",
3031 ProjectsEvent::Finished { .. } => "finished",
3032 }
3033 }
3034
3035 fn fields(&self) -> serde_json::Value {
3036 match self {
3037 ProjectsEvent::DesignersPoll { open } => serde_json::json!({ "open": open }),
3038 ProjectsEvent::DesignersProceed { sessions } => {
3039 serde_json::json!({ "sessions": sessions_json(sessions) })
3040 }
3041 ProjectsEvent::DesignersSkipped => serde_json::json!({ "reason": "record only" }),
3042 ProjectsEvent::RunStarted { run_id, action } => {
3043 serde_json::json!({ "run_id": run_id, "action": action })
3044 }
3045 ProjectsEvent::PreImageExported { bytes } => serde_json::json!({ "bytes": bytes }),
3046 ProjectsEvent::TagsPreImageExported { provider, bytes } => {
3047 serde_json::json!({ "provider": provider, "bytes": bytes })
3048 }
3049 ProjectsEvent::TagImportSent {
3050 provider,
3051 path,
3052 policy,
3053 nodes,
3054 carry,
3055 } => serde_json::json!({
3056 "provider": provider, "path": path, "policy": policy, "nodes": nodes,
3057 "carry": carry
3058 }),
3059 ProjectsEvent::TagValuesCarried {
3060 provider,
3061 path,
3062 nodes,
3063 } => serde_json::json!({ "provider": provider, "path": path, "nodes": nodes }),
3064 ProjectsEvent::TagProviderSettingsApplied { provider, action } => {
3065 serde_json::json!({ "provider": provider, "action": action })
3066 }
3067 ProjectsEvent::TagGroupsWritten {
3068 providers,
3069 written,
3070 deleted,
3071 } => serde_json::json!({
3072 "providers": providers, "written": written, "deleted": deleted
3073 }),
3074 ProjectsEvent::TagImportAnswered { provider, clean } => {
3075 serde_json::json!({ "provider": provider, "clean": clean })
3076 }
3077 ProjectsEvent::TagsVerified { provider, equal } => {
3078 serde_json::json!({ "provider": provider, "equal": equal })
3079 }
3080 ProjectsEvent::ImportSent { overwrite } => {
3081 serde_json::json!({ "overwrite": overwrite })
3082 }
3083 ProjectsEvent::Verified { equal } => serde_json::json!({ "equal": equal }),
3084 ProjectsEvent::RecordFailed { code } => serde_json::json!({ "code": code }),
3085 ProjectsEvent::TagRecordWritten { provider } => {
3086 serde_json::json!({ "provider": provider })
3087 }
3088 ProjectsEvent::TagRecordFailed { provider, code } => {
3089 serde_json::json!({ "provider": provider, "code": code })
3090 }
3091 ProjectsEvent::ChildrenSettling { children } => {
3092 serde_json::json!({ "children": children })
3093 }
3094 ProjectsEvent::ChildChecked { name, status } => {
3095 serde_json::json!({ "name": name, "status": status.as_str() })
3096 }
3097 ProjectsEvent::RollbackStarted { cause } => {
3098 serde_json::json!({ "cause": cause.as_str() })
3099 }
3100 ProjectsEvent::ProjectRestored { equal }
3101 | ProjectsEvent::RollbackVerified { equal } => {
3102 serde_json::json!({ "equal": equal })
3103 }
3104 ProjectsEvent::TagGroupsRestored { providers } => {
3105 serde_json::json!({ "providers": providers })
3106 }
3107 ProjectsEvent::TagProviderRestored { provider, action } => {
3108 serde_json::json!({ "provider": provider, "action": action })
3109 }
3110 ProjectsEvent::TagDeleteSent {
3111 provider,
3112 paths,
3113 rollback,
3114 } => serde_json::json!({ "provider": provider, "paths": paths, "rollback": rollback }),
3115 ProjectsEvent::TagDeleteAnswered {
3116 provider,
3117 clean,
3118 deleted,
3119 } => serde_json::json!({ "provider": provider, "clean": clean, "deleted": deleted }),
3120 ProjectsEvent::HostPreImageWritten { files } => serde_json::json!({ "files": files }),
3121 ProjectsEvent::JournalWritten { state } => serde_json::json!({ "state": state }),
3122 ProjectsEvent::BackupWaitStarted { budget_s } => {
3123 serde_json::json!({ "budget_s": budget_s })
3124 }
3125 ProjectsEvent::BackupResync { ok, code } => {
3126 serde_json::json!({ "ok": ok, "code": code })
3127 }
3128 ProjectsEvent::BackupVerdict { converged, reason } => {
3129 serde_json::json!({ "converged": converged, "reason": reason })
3130 }
3131 ProjectsEvent::Finished { outcome } => {
3132 serde_json::json!({ "outcome": outcome.as_str() })
3133 }
3134 ProjectsEvent::ImportAnswered
3135 | ProjectsEvent::RecordWritten
3136 | ProjectsEvent::JournalCleared => serde_json::Value::Null,
3137 }
3138 }
3139}
3140
3141pub trait ProjectsObserver: Send + Sync {
3144 fn event(&self, event: &ProjectsEvent);
3146}
3147
3148pub struct NoProjectsObserver;
3150
3151impl ProjectsObserver for NoProjectsObserver {
3152 fn event(&self, _event: &ProjectsEvent) {}
3153}
3154
3155pub struct ProjectsRunOptions<'a> {
3157 pub repo: PathBuf,
3159 pub targets_root: PathBuf,
3162 pub target_file: String,
3164 pub master: NodeTarget,
3168 #[doc(hidden)]
3171 pub record_writer: Option<&'a dyn ReleaseRecordWriter>,
3172 #[doc(hidden)]
3175 pub tag_record_writer: Option<&'a dyn TagRecordWriter>,
3176 #[doc(hidden)]
3180 pub run_writer: Option<&'a dyn ProjectRunWriter>,
3181 pub designers: DesignersPolicy,
3183 pub children_settle: Duration,
3186 pub children_poll: Duration,
3190 pub tag_verify_settle: Duration,
3193 pub tag_verify_poll: Duration,
3196 pub backup: Option<&'a dyn BackupTags>,
3200 pub backup_policy: BackupPolicy,
3203 pub tag_delete_ready: Duration,
3208 pub tag_delete_poll: Duration,
3211}
3212
3213struct RunEvents<'a> {
3217 log: EventLog,
3218 observer: &'a dyn ProjectsObserver,
3219 late_failure: bool,
3220}
3221
3222impl RunEvents<'_> {
3223 fn emit(&mut self, event: ProjectsEvent) -> Result<(), CoreError> {
3224 self.observer.event(&event);
3225 self.log.append(event.name(), event.fields())
3226 }
3227
3228 fn record(&mut self, event: &ProjectsEvent) -> Result<(), CoreError> {
3230 self.log.append(event.name(), event.fields())
3231 }
3232
3233 fn emit_late(&mut self, event: ProjectsEvent) {
3234 if self.emit(event).is_err() {
3235 self.late_failure = true;
3236 }
3237 }
3238}
3239
3240fn sessions_json(sessions: &[DesignerSeen]) -> Vec<serde_json::Value> {
3243 sessions
3244 .iter()
3245 .map(|(user, project, host)| {
3246 serde_json::json!({ "user": user, "project": project, "host": host })
3247 })
3248 .collect()
3249}
3250
3251struct DesignersFound {
3253 mode: &'static str,
3256 timeout_s: Option<u64>,
3258 polls: Vec<usize>,
3260 seen: Vec<DesignerSeen>,
3262 proceeded: bool,
3264 skipped: bool,
3266}
3267
3268async fn designers_check(
3273 plan: &DeployProjectsResult,
3274 gateway: &dyn ProjectGateway,
3275 policy: &DesignersPolicy,
3276 observer: &dyn ProjectsObserver,
3277) -> Result<DesignersFound, CoreError> {
3278 if matches!(plan.action, ProjectAction::Record | ProjectAction::Skip) {
3281 return Ok(DesignersFound {
3282 mode: "skipped",
3283 timeout_s: None,
3284 polls: Vec::new(),
3285 seen: Vec::new(),
3286 proceeded: false,
3287 skipped: true,
3288 });
3289 }
3290 let mut polls = Vec::new();
3291 let (verdict, seen) =
3292 designers_gate(|| gateway.designers(), policy, &SystemClock, &mut |open| {
3293 polls.push(open);
3294 observer.event(&ProjectsEvent::DesignersPoll { open });
3295 Ok(())
3296 })
3297 .await?;
3298 if verdict == DesignersVerdict::Open {
3299 return Err(CoreError::DeployPreflightFailed {
3300 failures: Box::new([PreflightFailure {
3301 check: "designers".to_string(),
3302 detail: designers_detail(&seen),
3303 }]),
3304 });
3305 }
3306 Ok(DesignersFound {
3307 mode: policy.mode.as_str(),
3308 timeout_s: Some(policy.timeout.as_secs()),
3309 polls,
3310 seen,
3311 proceeded: verdict == DesignersVerdict::Proceeded,
3312 skipped: false,
3313 })
3314}
3315
3316async fn wrong_target_check(
3321 loaded: &LoadedProject,
3322 gateway: &dyn ProjectGateway,
3323 opts: &ProjectsRunOptions<'_>,
3324) -> Result<(), CoreError> {
3325 let rest = gateway.gateway_name().await?;
3326 let reader = for_node(
3327 &opts.targets_root,
3328 &opts.target_file,
3329 "nodes.master",
3330 &opts.master,
3331 )?;
3332 let problems = wrong_target_gate(
3333 rest.as_deref(),
3334 reader.as_ref(),
3335 &loaded.target.mode,
3336 opts.master.gateway_name.as_deref(),
3337 )
3338 .await?;
3339 if problems.is_empty() {
3340 return Ok(());
3341 }
3342 Err(CoreError::DeployPreflightFailed {
3343 failures: Box::new([wrong_target_failure(&problems)]),
3344 })
3345}
3346
3347fn release_record(plan: &DeployProjectsResult, release: &str, run_id: &RunId) -> ReleaseRecord {
3349 ReleaseRecord {
3350 schema_version: 1,
3351 gateway: plan.gateway.clone(),
3352 project: plan.project.clone(),
3353 release: release.to_string(),
3354 artifact_sha256: plan.artifact_sha256.clone(),
3355 run_id: run_id.as_str().to_string(),
3356 deployed_at: rfc3339_now_utc(),
3357 }
3358}
3359
3360fn provider_not_allowed(providers: &[&str]) -> CoreError {
3364 CoreError::WebdevRouteError {
3365 code: "provider_not_allowed".to_string(),
3366 message: format!(
3367 "the tag-delete route does not allow tag provider(s) {}, which this run removes tags \
3368 from; reinstall the route from this project's repo with `ign webdev deploy \
3369 --with-tag-delete --tag-delete-repo <repo> --tag-delete-user-source <name> \
3370 --tag-delete-role <role>`; nothing was written",
3371 providers.join(", ")
3372 ),
3373 endpoint: None,
3374 }
3375}
3376
3377fn stale_live(expected: String, actual: String) -> CoreError {
3378 CoreError::PlanStale {
3379 expected,
3380 actual,
3381 detail: "live project changed since the plan".to_string(),
3382 }
3383}
3384
3385pub async fn deploy_projects_run(
3392 prepared: ProjectsPrepared,
3393 loaded: &LoadedProject,
3394 input: &ProjectsInput,
3395 surfaces: &ProjectsSurfaces<'_>,
3396 opts: &ProjectsRunOptions<'_>,
3397 observer: &dyn ProjectsObserver,
3398) -> Result<DeployProjectsResult, CoreError> {
3399 let planned = match &prepared {
3402 ProjectsPrepared::Unchanged(result) => result,
3403 ProjectsPrepared::Ready { plan, .. } => plan,
3404 };
3405 if let Some(journal) = planned.tags.as_ref().and_then(|t| t.journal.as_ref()) {
3406 return Err(project_journal_present(loaded, &input.env, &journal.run_id));
3407 }
3408 let (mut plan, tag_records) = match prepared {
3409 ProjectsPrepared::Unchanged(result) => return Ok(result),
3410 ProjectsPrepared::Ready {
3411 plan, tag_records, ..
3412 } => (plan, tag_records),
3413 };
3414 let removals = plan
3424 .tags
3425 .as_ref()
3426 .is_some_and(|tags| tags.removed_count > 0);
3427 let group_removals: Vec<String> = plan
3428 .tags
3429 .as_ref()
3430 .map(|tags| {
3431 tags.groups
3432 .iter()
3433 .flat_map(|g| {
3434 g.removed
3435 .iter()
3436 .map(|name| format!("tag group {}", full_tag_path(&g.provider, name)))
3437 })
3438 .collect()
3439 })
3440 .unwrap_or_default();
3441 if let Some(tags) = &plan.tags
3442 && (removals || !group_removals.is_empty())
3443 && !input.allow_destructive
3444 {
3445 let mut keys = tags.removed.clone();
3446 keys.extend(group_removals.iter().cloned());
3447 return Err(CoreError::DestructiveChangeRefused {
3448 gate: "tag_removal".to_string(),
3449 keys,
3450 count: tags.removed_count + group_removals.len(),
3451 });
3452 }
3453 if removals && let TagDeleteSurface::NotConfigured { missing } = &surfaces.tag_delete {
3456 return Err(CoreError::TagDeleteNotConfigured {
3457 profile: loaded.target.nodes.master.profile.clone(),
3458 missing: missing.clone(),
3459 });
3460 }
3461 let built;
3464 let writer: &dyn ReleaseRecordWriter = match opts.record_writer {
3465 Some(writer) => writer,
3466 None => {
3467 built = release_writer_for_node(
3468 &opts.targets_root,
3469 &opts.target_file,
3470 "nodes.master",
3471 &opts.master,
3472 )?;
3473 built.as_ref()
3474 }
3475 };
3476 let built_tags;
3477 let tag_writer: Option<&dyn TagRecordWriter> = match (&plan.tags, opts.tag_record_writer) {
3478 (None, _) => None,
3479 (Some(_), Some(writer)) => Some(writer),
3480 (Some(_), None) => {
3481 built_tags = tag_record_writer_for_node(
3482 &opts.targets_root,
3483 &opts.target_file,
3484 "nodes.master",
3485 &opts.master,
3486 )?;
3487 Some(built_tags.as_ref())
3488 }
3489 };
3490 let built_run;
3492 let run_writer: Option<&dyn ProjectRunWriter> = match (&plan.tags, opts.run_writer) {
3493 (None, _) => None,
3494 (Some(_), Some(writer)) => Some(writer),
3495 (Some(_), None) => {
3496 built_run = project_run_writer_for_node(
3497 &opts.targets_root,
3498 &opts.target_file,
3499 "nodes.master",
3500 &opts.master,
3501 )?;
3502 Some(built_run.as_ref())
3503 }
3504 };
3505 let built_host;
3510 let group_writes = plan
3511 .tags
3512 .as_ref()
3513 .is_some_and(|tags| tags.groups.iter().any(GroupPlan::would_change));
3514 let apply_host: Option<&dyn ApplyHost> = match surfaces.apply_host {
3515 Some(host) => Some(host),
3516 None if group_writes => {
3517 built_host = apply_host_for_node(
3518 &opts.targets_root,
3519 &opts.target_file,
3520 "nodes.master",
3521 &opts.master,
3522 &loaded.target.gateway,
3523 &loaded.target.mode,
3524 )?;
3525 Some(built_host.as_ref())
3526 }
3527 None => None,
3528 };
3529 let surfaces = &ProjectsSurfaces {
3530 gateway: surfaces.gateway,
3531 host: surfaces.host,
3532 tree: surfaces.tree,
3533 apply_host,
3534 deploy: surfaces.deploy,
3535 tag_delete: surfaces.tag_delete.clone(),
3536 };
3537 wrong_target_check(loaded, surfaces.gateway, opts).await?;
3540 let designers = designers_check(&plan, surfaces.gateway, &opts.designers, observer).await?;
3541 let mut answer = match (&plan.tags, &surfaces.tag_delete) {
3550 (Some(_), TagDeleteSurface::Configured(route)) => {
3551 Some(wait_ready(*route, opts.tag_delete_ready, opts.tag_delete_poll).await)
3552 }
3553 _ => None,
3554 };
3555 if removals {
3556 if let Some(Err(_)) = &answer
3557 && let Some(Err(err)) = answer.take()
3558 {
3559 return Err(err);
3560 }
3561 if let (Some(Ok(handshake)), Some(tags)) = (&answer, &plan.tags) {
3562 let refused: Vec<&str> = tags
3563 .providers
3564 .iter()
3565 .filter(|p| p.removed > 0)
3566 .map(|p| p.provider.as_str())
3567 .filter(|name| !handshake.providers.iter().any(|p| p == name))
3568 .collect();
3569 if !refused.is_empty() {
3570 return Err(provider_not_allowed(&refused));
3571 }
3572 }
3573 }
3574 let route = RouteState::of(&surfaces.tag_delete, answer);
3575 if let Some(tags) = plan.tags.as_mut() {
3576 tags.delete_route = Some(route.report(&owned_names(loaded)));
3577 if let (Some(_), None) = (route.route, &route.handshake) {
3578 tags.warnings.push(format!(
3579 "the tag-delete route is not ready ({}); a rollback cannot remove tags this run \
3580 creates",
3581 route.problem.as_deref().unwrap_or("no handshake")
3582 ));
3583 }
3584 }
3585 let (run_id, run_dir) = RunDir::create_fresh(&opts.repo, &plan.gateway)?;
3586 let rel = run_dir_rel(&plan.gateway, &run_id);
3587 let outcome = execute(
3588 &plan,
3589 &tag_records,
3590 loaded,
3591 input,
3592 surfaces,
3593 opts,
3594 Writers {
3595 release: writer,
3596 tags: tag_writer,
3597 run: run_writer,
3598 },
3599 &run_dir,
3600 &run_id,
3601 &rel,
3602 &designers,
3603 &route,
3604 observer,
3605 )
3606 .await;
3607 match outcome {
3608 Ok(mut result) => {
3609 if let Err(err) = run_dir.write_json("result.json", &result) {
3610 result.warnings.push(format!(
3611 "the run's result.json could not be written ({})",
3612 err.code()
3613 ));
3614 }
3615 Ok(result)
3616 }
3617 Err(err) => {
3618 let _ = run_dir.write_json(
3620 "result.json",
3621 &serde_json::json!({
3622 "ok": false,
3623 "slug": err.code(),
3624 "run_id": run_id.as_str(),
3625 }),
3626 );
3627 Err(err)
3628 }
3629 }
3630}
3631
3632struct ChildSweep {
3634 children: Vec<ChildReport>,
3635 warnings: Vec<String>,
3636}
3637
3638type ChildRead = Result<String, String>;
3641
3642async fn read_children(gateway: &dyn ProjectGateway, names: &[String]) -> Vec<ChildRead> {
3643 let mut reads = Vec::with_capacity(names.len());
3644 for name in names {
3645 reads.push(match gateway.export_bytes(name).await {
3646 Ok(zip) => live_digest(name, &zip)
3647 .map(|digest| digest.sha256)
3648 .map_err(|err| err.code().to_string()),
3649 Err(err) => Err(err.code().to_string()),
3650 });
3651 }
3652 reads
3653}
3654
3655async fn sweep_children(
3666 result: &DeployProjectsResult,
3667 loaded: &LoadedProject,
3668 surfaces: &ProjectsSurfaces<'_>,
3669 opts: &ProjectsRunOptions<'_>,
3670 events: &mut RunEvents<'_>,
3671) -> ChildSweep {
3672 let project = result.project.as_str();
3673 let gateway = surfaces.gateway;
3674 let mut sweep = ChildSweep {
3675 children: Vec::new(),
3676 warnings: Vec::new(),
3677 };
3678
3679 let records = match gateway.projects().await {
3680 Ok(records) => records,
3681 Err(err) => {
3682 sweep.warnings.push(format!(
3683 "could not list the projects to check the children of {project} ({})",
3684 err.code()
3685 ));
3686 return sweep;
3687 }
3688 };
3689 let mut names: Vec<String> = records
3690 .into_iter()
3691 .filter(|record| record.parent.as_deref() == Some(project))
3692 .map(|record| record.name)
3693 .collect();
3694 names.sort_by(|a, b| a.as_bytes().cmp(b.as_bytes()));
3695 names.dedup();
3696 if names.is_empty() {
3697 return sweep;
3698 }
3699 events.emit_late(ProjectsEvent::ChildrenSettling {
3700 children: names.clone(),
3701 });
3702
3703 let started = Instant::now();
3706 let mut current = read_children(gateway, &names).await;
3707 let mut unsettled = vec![true; names.len()];
3708 loop {
3709 let elapsed = started.elapsed();
3710 if elapsed >= opts.children_settle {
3711 break;
3712 }
3713 tokio::time::sleep(opts.children_poll.min(opts.children_settle - elapsed)).await;
3714 let next = read_children(gateway, &names).await;
3715 unsettled = current.iter().zip(&next).map(|(a, b)| a != b).collect();
3716 current = next;
3717 if !unsettled.iter().any(|changing| *changing) {
3718 break;
3719 }
3720 }
3721
3722 let gateway_name = result.gateway.as_str();
3723 let root = LedgerRoot::for_node(&loaded.target.nodes.master);
3724 for ((name, read), changing) in names.iter().zip(current).zip(unsettled) {
3725 let live = match read {
3726 Ok(sha256) => Some(sha256),
3727 Err(code) => {
3728 sweep.warnings.push(format!(
3729 "child project {name}: its export could not be read ({code})"
3730 ));
3731 None
3732 }
3733 };
3734 if changing {
3735 sweep.warnings.push(format!(
3736 "child project {name} did not settle within {} s",
3737 opts.children_settle.as_secs()
3738 ));
3739 }
3740 let record = match surfaces.host.read_release_record(gateway_name, name).await {
3741 Ok(None) => None,
3742 Ok(Some(bytes)) => {
3743 let location = release_record_location(root, gateway_name, name);
3744 match ReleaseRecord::parse(&bytes, &location, gateway_name, name) {
3745 Ok(record) => Some(record),
3746 Err(err) => {
3747 sweep.warnings.push(format!(
3748 "child project {name}: its release record could not be read ({})",
3749 err.code()
3750 ));
3751 None
3752 }
3753 }
3754 }
3755 Err(err) => {
3756 sweep.warnings.push(format!(
3757 "child project {name}: its release record could not be read ({})",
3758 err.code()
3759 ));
3760 None
3761 }
3762 };
3763 let status = match (&record, &live) {
3764 (Some(record), Some(live)) if record.artifact_sha256 == *live => ChildStatus::Ok,
3765 (Some(record), Some(_)) => {
3766 sweep.warnings.push(format!(
3767 "child project {name} no longer matches the release this tool last deployed to it ({})",
3768 record.release
3769 ));
3770 ChildStatus::Mismatch
3771 }
3772 _ => ChildStatus::Unverified,
3773 };
3774 events.emit_late(ProjectsEvent::ChildChecked {
3775 name: name.clone(),
3776 status,
3777 });
3778 sweep.children.push(ChildReport {
3779 name: name.clone(),
3780 status,
3781 release: record.as_ref().map(|r| r.release.clone()),
3782 recorded_sha256: record.map(|r| r.artifact_sha256),
3783 live_sha256: live,
3784 });
3785 }
3786 sweep
3787}
3788
3789#[allow(clippy::too_many_arguments)]
3790async fn execute(
3791 plan: &DeployProjectsResult,
3792 tag_records: &std::collections::BTreeMap<String, TagRecord>,
3793 loaded: &LoadedProject,
3794 input: &ProjectsInput,
3795 surfaces: &ProjectsSurfaces<'_>,
3796 opts: &ProjectsRunOptions<'_>,
3797 writers: Writers<'_>,
3798 run_dir: &RunDir,
3799 run_id: &RunId,
3800 rel: &str,
3801 designers: &DesignersFound,
3802 route: &RouteState<'_>,
3803 observer: &dyn ProjectsObserver,
3804) -> Result<DeployProjectsResult, CoreError> {
3805 let project = plan.project.as_str();
3806 let gateway = surfaces.gateway;
3807 let artifact = &loaded.artifact.digest;
3808 let writer = writers.release;
3809
3810 run_dir.write_json(
3811 "run.json",
3812 &serde_json::json!({
3813 "run_id": run_id.as_str(),
3814 "started_at": rfc3339_now_utc(),
3815 "env": plan.env,
3816 "gateway": plan.gateway,
3817 "project": plan.project,
3818 "release": plan.release,
3819 "action": plan.action,
3820 "artifact_sha256": plan.artifact_sha256,
3821 "live_sha256": plan.live_sha256,
3822 "allow_downgrade": input.allow_downgrade,
3823 "allow_first_change": input.allow_first_change,
3824 "tag_providers": loaded.file.tag_providers,
3825 "designers": designers.mode,
3826 "designers_timeout_s": designers.timeout_s,
3827 "designers_seen": sessions_json(&designers.seen),
3828 "run_dir_git": run_dir.git_status().as_record(),
3829 }),
3830 )?;
3831 run_dir.write_json("plan.json", plan)?;
3832 let mut events = RunEvents {
3833 log: EventLog::open(run_dir)?,
3834 observer,
3835 late_failure: false,
3836 };
3837 events.emit(ProjectsEvent::RunStarted {
3838 run_id: run_id.as_str().to_string(),
3839 action: plan.action,
3840 })?;
3841 for open in &designers.polls {
3844 events.record(&ProjectsEvent::DesignersPoll { open: *open })?;
3845 }
3846 if designers.proceeded {
3847 events.emit(ProjectsEvent::DesignersProceed {
3848 sessions: designers.seen.clone(),
3849 })?;
3850 }
3851 if designers.skipped {
3852 events.emit(ProjectsEvent::DesignersSkipped)?;
3853 }
3854
3855 let mut result = plan.clone();
3856 result.run_id = Some(run_id.as_str().to_string());
3857 result.run_dir = Some(rel.to_string());
3858 let record = release_record(plan, &loaded.release.tag, run_id);
3859 let tag_owning = plan.tags.is_some();
3860 let backup_since_ms = wall_clock_ms();
3863 let owned_names = owned_names(loaded);
3864
3865 if plan.action == ProjectAction::Record {
3866 let zip = gateway.export_bytes(project).await?;
3869 let live = live_digest(project, &zip)?;
3870 if live.sha256 != artifact.sha256 {
3871 return Err(stale_live(artifact.sha256.clone(), live.sha256));
3872 }
3873 events.emit(ProjectsEvent::Verified { equal: true })?;
3874 let step = match deploy_owned_tags(
3877 plan,
3878 tag_records,
3879 loaded,
3880 input,
3881 surfaces,
3882 opts,
3883 run_dir,
3884 run_id,
3885 HostWrites {
3886 writer: writers.run,
3887 project_zip: None,
3888 },
3889 route,
3890 &mut events,
3891 )
3892 .await
3893 {
3894 Ok(step) => step,
3895 Err(TagStepError::Unwritten(err)) => return Err(err),
3896 Err(TagStepError::Written(failure)) => {
3897 return roll_back(
3898 *failure,
3899 None,
3900 result,
3901 loaded,
3902 input,
3903 surfaces,
3904 opts,
3905 run_id,
3906 rel,
3907 UndoWriters {
3908 run: writers.run,
3909 records: None,
3910 release: None,
3911 },
3912 route,
3913 &mut events,
3914 )
3915 .await;
3916 }
3917 };
3918 note_tag_step(&mut result, &step);
3919 let changed = step.writes.wrote();
3924 let mut record_written = true;
3925 if let Err(err) = writer
3926 .write_release_record(&plan.gateway, project, &record.to_bytes())
3927 .await
3928 {
3929 events.emit_late(ProjectsEvent::RecordFailed {
3930 code: err.code().to_string(),
3931 });
3932 if !changed {
3933 return Err(unwritten(
3934 err,
3935 loaded,
3936 writers.run,
3937 step.writes.host.as_ref(),
3938 &mut events,
3939 )
3940 .await);
3941 }
3942 record_written = false;
3943 result.warnings.push(format!(
3944 "the tags are verified but the release record could not be written ({})",
3945 err.code()
3946 ));
3947 }
3948 if let Err(err) = write_tag_records(
3951 &mut result,
3952 loaded,
3953 run_id,
3954 writers.tags,
3955 &mut events,
3956 step.batches_sent + step.other_writes > 0,
3957 )
3958 .await
3959 {
3960 if changed {
3961 return Err(err);
3962 }
3963 return Err(unwritten(
3964 err,
3965 loaded,
3966 writers.run,
3967 step.writes.host.as_ref(),
3968 &mut events,
3969 )
3970 .await);
3971 }
3972 clear_host_journal(
3974 &mut result,
3975 loaded,
3976 writers.run,
3977 step.writes.host.as_ref(),
3978 &mut events,
3979 )
3980 .await;
3981 let backup = backup_step(
3983 &mut result,
3984 &owned_names,
3985 surfaces,
3986 opts,
3987 backup_since_ms,
3988 &mut events,
3989 )
3990 .await;
3991 finish_recorded(
3995 &mut result,
3996 &mut events,
3997 backup == Some(false),
3998 record_written,
3999 );
4000 return Ok(result);
4001 }
4002
4003 let overwrite = plan.action == ProjectAction::Import;
4004 let mut pre_zip: Option<Vec<u8>> = None;
4007 if overwrite {
4008 let pre = gateway.export_bytes(project).await?;
4012 let live = live_digest(project, &pre)?;
4013 let mut file = run_dir.create_private_file_in("export", &format!("{project}.zip"))?;
4014 file.write_all(&pre)
4015 .and_then(|()| file.sync_all())
4016 .map_err(|e| CoreError::DeployRunInvalid {
4017 run_id: run_id.as_str().to_string(),
4018 problem: format!("could not write the pre-image export ({:?})", e.kind()),
4019 })?;
4020 events.emit(ProjectsEvent::PreImageExported { bytes: pre.len() })?;
4021 if plan.live_sha256.as_deref() != Some(live.sha256.as_str()) {
4024 return Err(stale_live(
4025 plan.live_sha256
4026 .clone()
4027 .unwrap_or_else(|| "none".to_string()),
4028 live.sha256,
4029 ));
4030 }
4031 pre_zip = Some(pre);
4032 }
4033
4034 let step = match deploy_owned_tags(
4037 plan,
4038 tag_records,
4039 loaded,
4040 input,
4041 surfaces,
4042 opts,
4043 run_dir,
4044 run_id,
4045 HostWrites {
4046 writer: writers.run,
4047 project_zip: pre_zip.as_deref(),
4048 },
4049 route,
4050 &mut events,
4051 )
4052 .await
4053 {
4054 Ok(step) => step,
4055 Err(TagStepError::Unwritten(err)) => return Err(err),
4056 Err(TagStepError::Written(failure)) => {
4057 return roll_back(
4058 *failure,
4059 None,
4060 result,
4061 loaded,
4062 input,
4063 surfaces,
4064 opts,
4065 run_id,
4066 rel,
4067 UndoWriters {
4068 run: writers.run,
4069 records: None,
4070 release: None,
4071 },
4072 route,
4073 &mut events,
4074 )
4075 .await;
4076 }
4077 };
4078 note_tag_step(&mut result, &step);
4079 let tags_wrote = step.writes.wrote();
4082 if plan.action == ProjectAction::Skip {
4083 if let Err(err) = write_tag_records(
4088 &mut result,
4089 loaded,
4090 run_id,
4091 writers.tags,
4092 &mut events,
4093 step.batches_sent + step.other_writes > 0,
4094 )
4095 .await
4096 {
4097 if tags_wrote {
4098 return Err(err);
4099 }
4100 return Err(unwritten(
4101 err,
4102 loaded,
4103 writers.run,
4104 step.writes.host.as_ref(),
4105 &mut events,
4106 )
4107 .await);
4108 }
4109 clear_host_journal(
4110 &mut result,
4111 loaded,
4112 writers.run,
4113 step.writes.host.as_ref(),
4114 &mut events,
4115 )
4116 .await;
4117 result.outcome = ProjectOutcome::Applied;
4118 if backup_step(
4119 &mut result,
4120 &owned_names,
4121 surfaces,
4122 opts,
4123 backup_since_ms,
4124 &mut events,
4125 )
4126 .await
4127 == Some(false)
4128 {
4129 mark_backup_unsynced(&mut result);
4130 }
4131 events.emit_late(ProjectsEvent::Finished {
4132 outcome: result.outcome,
4133 });
4134 note_late_log_failure(&mut result, &events);
4135 return Ok(result);
4136 }
4137
4138 if tag_owning && !designers.skipped {
4143 match designers_check(plan, gateway, &opts.designers, observer).await {
4144 Ok(found) => {
4145 if found.proceeded {
4146 events.emit_late(ProjectsEvent::DesignersProceed {
4147 sessions: found.seen.clone(),
4148 });
4149 }
4150 }
4151 Err(err) if tags_wrote => {
4152 let failure = TagFailure {
4153 cause: RollbackCause::ProjectImportFailed,
4154 what: format!(
4155 "the Designer check right before the project import did not pass ({err})"
4156 ),
4157 writes: step.writes,
4158 };
4159 return roll_back(
4160 failure,
4161 Some(ProjectUndo { pre: pre_zip }),
4162 result,
4163 loaded,
4164 input,
4165 surfaces,
4166 opts,
4167 run_id,
4168 rel,
4169 UndoWriters {
4170 run: writers.run,
4171 records: None,
4172 release: None,
4173 },
4174 route,
4175 &mut events,
4176 )
4177 .await;
4178 }
4179 Err(err) => {
4180 return Err(unwritten(
4181 err,
4182 loaded,
4183 writers.run,
4184 step.writes.host.as_ref(),
4185 &mut events,
4186 )
4187 .await);
4188 }
4189 }
4190 }
4191
4192 if tag_owning {
4195 events.emit_late(ProjectsEvent::ImportSent { overwrite });
4196 } else {
4197 events.emit(ProjectsEvent::ImportSent { overwrite })?;
4198 }
4199 let answer = match gateway
4200 .project_import(project, loaded.artifact.zip.clone(), overwrite)
4201 .await
4202 {
4203 Ok(answer) => Some(answer),
4204 Err(err) if import_outcome_unknown(&err) => {
4209 result.warnings.push(format!(
4210 "the import request failed after it may have reached the gateway ({}); the project was read back to find out what it holds",
4211 err.code()
4212 ));
4213 None
4214 }
4215 Err(err) if tags_wrote => {
4219 let failure = TagFailure {
4220 cause: RollbackCause::ProjectImportFailed,
4221 what: format!("the project import was refused ({})", err.code()),
4222 writes: step.writes,
4223 };
4224 return roll_back(
4225 failure,
4226 Some(ProjectUndo { pre: pre_zip }),
4227 result,
4228 loaded,
4229 input,
4230 surfaces,
4231 opts,
4232 run_id,
4233 rel,
4234 UndoWriters {
4235 run: writers.run,
4236 records: None,
4237 release: None,
4238 },
4239 route,
4240 &mut events,
4241 )
4242 .await;
4243 }
4244 Err(err) => {
4245 return Err(unwritten(
4246 err,
4247 loaded,
4248 writers.run,
4249 step.writes.host.as_ref(),
4250 &mut events,
4251 )
4252 .await);
4253 }
4254 };
4255 if let Some(answer) = &answer {
4258 events.emit_late(ProjectsEvent::ImportAnswered);
4259 if answer
4260 .response
4261 .get("changes")
4262 .and_then(serde_json::Value::as_array)
4263 .is_some_and(Vec::is_empty)
4264 {
4265 result
4266 .warnings
4267 .push("the gateway reported no changes for this import".to_string());
4268 }
4269 }
4270
4271 let equal = match retry_transient(|| gateway.export_bytes(project)).await {
4272 Ok(zip) => match live_digest(project, &zip) {
4273 Ok(live) => Some(live.sha256 == artifact.sha256),
4274 Err(err) => {
4275 result.warnings.push(format!(
4276 "could not verify the import: the project read back is unusable ({})",
4277 err.code()
4278 ));
4279 None
4280 }
4281 },
4282 Err(err) => {
4283 result.warnings.push(format!(
4284 "could not verify the import: the project could not be read back ({})",
4285 err.code()
4286 ));
4287 None
4288 }
4289 };
4290 events.emit_late(ProjectsEvent::Verified {
4291 equal: equal == Some(true),
4292 });
4293
4294 if equal == Some(true) {
4295 match writer
4296 .write_release_record(&plan.gateway, project, &record.to_bytes())
4297 .await
4298 {
4299 Ok(()) => {
4300 events.emit_late(ProjectsEvent::RecordWritten);
4301 result.record_written = true;
4302 }
4303 Err(err) => {
4304 events.emit_late(ProjectsEvent::RecordFailed {
4305 code: err.code().to_string(),
4306 });
4307 result.warnings.push(format!(
4308 "the import is verified but the release record could not be written ({})",
4309 err.code()
4310 ));
4311 }
4312 }
4313 let _ =
4316 write_tag_records(&mut result, loaded, run_id, writers.tags, &mut events, true).await;
4317 clear_host_journal(
4320 &mut result,
4321 loaded,
4322 writers.run,
4323 step.writes.host.as_ref(),
4324 &mut events,
4325 )
4326 .await;
4327 if let Some(parent) = &loaded.meta.parent {
4330 match gateway.projects().await {
4331 Ok(records) => {
4332 if records
4333 .iter()
4334 .find(|r| r.name == project)
4335 .is_some_and(|r| r.invalid_parent == Some(true))
4336 {
4337 result.warnings.push(format!(
4338 "project {project}: after the import the gateway reports that its parent {parent} is invalid"
4339 ));
4340 }
4341 }
4342 Err(err) => result.warnings.push(format!(
4343 "project {project}: the parent could not be read back after the import ({})",
4344 err.code()
4345 )),
4346 }
4347 }
4348 result.outcome = if tag_owning {
4349 ProjectOutcome::Applied
4350 } else if overwrite {
4351 ProjectOutcome::Imported
4352 } else {
4353 ProjectOutcome::Created
4354 };
4355 } else if tags_wrote {
4356 let failure = TagFailure {
4359 cause: if answer.is_none() {
4360 RollbackCause::ProjectImportFailed
4361 } else {
4362 RollbackCause::ProjectUnverified
4363 },
4364 what: if equal == Some(false) {
4365 "the project read back does not equal the artifact".to_string()
4366 } else {
4367 "the project could not be read back or compared after its import".to_string()
4368 },
4369 writes: step.writes,
4370 };
4371 return roll_back(
4372 failure,
4373 Some(ProjectUndo { pre: pre_zip }),
4374 result,
4375 loaded,
4376 input,
4377 surfaces,
4378 opts,
4379 run_id,
4380 rel,
4381 UndoWriters {
4382 run: writers.run,
4383 records: None,
4384 release: None,
4385 },
4386 route,
4387 &mut events,
4388 )
4389 .await;
4390 } else {
4391 result.outcome = ProjectOutcome::ImportedUnverified;
4394 result.ok = false;
4395 result.slug = Some(PROJECT_UNVERIFIED_SLUG.to_string());
4396 if overwrite {
4400 result.restore_command = Some(restore_command(project, rel));
4401 }
4402 if step.writes.host.is_some() {
4405 result.warnings.push(format!(
4406 "the host journal {} stays; `{}` restores the project from the host copy",
4407 host_journal_location(loaded),
4408 rollback_run_command(&input.env, run_id.as_str())
4409 ));
4410 }
4411 }
4412 let swept = match result.outcome {
4416 ProjectOutcome::Imported | ProjectOutcome::ImportedUnverified => true,
4417 ProjectOutcome::Applied => overwrite,
4418 _ => false,
4419 };
4420 if swept {
4421 let sweep = sweep_children(&result, loaded, surfaces, opts, &mut events).await;
4422 result.children = sweep.children;
4423 result.warnings.extend(sweep.warnings);
4424 }
4425 if result.outcome == ProjectOutcome::Applied
4428 && backup_step(
4429 &mut result,
4430 &owned_names,
4431 surfaces,
4432 opts,
4433 backup_since_ms,
4434 &mut events,
4435 )
4436 .await
4437 == Some(false)
4438 {
4439 mark_backup_unsynced(&mut result);
4440 }
4441 events.emit_late(ProjectsEvent::Finished {
4442 outcome: result.outcome,
4443 });
4444 note_late_log_failure(&mut result, &events);
4445 Ok(result)
4446}
4447
4448const TAG_FAILURE_PATHS_SHOWN: usize = 20;
4450
4451fn full_tag_path(provider: &str, path: &str) -> String {
4453 format!("[{provider}]{path}")
4454}
4455
4456struct TagStep {
4458 carried: Vec<String>,
4460 verify: Option<TagVerify>,
4462 batches_sent: usize,
4464 other_writes: usize,
4467 writes: TagWrites,
4470 deleted: Vec<String>,
4472 absent: Vec<String>,
4475}
4476
4477#[derive(Clone, Copy)]
4480struct Writers<'a> {
4481 release: &'a dyn ReleaseRecordWriter,
4482 tags: Option<&'a dyn TagRecordWriter>,
4483 run: Option<&'a dyn ProjectRunWriter>,
4485}
4486
4487type ProviderReads = BTreeMap<String, BTreeMap<String, Value>>;
4489
4490async fn read_providers(
4492 gateway: &dyn ProjectGateway,
4493 names: &[&str],
4494) -> Result<ProviderReads, CoreError> {
4495 let mut out = BTreeMap::new();
4496 for name in names {
4497 out.insert((*name).to_string(), read_flat_once(gateway, name).await?);
4498 }
4499 Ok(out)
4500}
4501
4502async fn read_flat(
4504 gateway: &dyn ProjectGateway,
4505 name: &str,
4506) -> Result<BTreeMap<String, Value>, CoreError> {
4507 let bytes = retry_transient(|| gateway.tag_export(name)).await?;
4508 flatten_export(&bytes, name)
4509}
4510
4511async fn read_flat_once(
4514 gateway: &dyn ProjectGateway,
4515 name: &str,
4516) -> Result<BTreeMap<String, Value>, CoreError> {
4517 let bytes = gateway.tag_export(name).await?;
4518 flatten_export(&bytes, name)
4519}
4520
4521fn reads_agree(a: &ProviderReads, b: &ProviderReads) -> bool {
4524 a.len() == b.len()
4525 && a.iter().all(|(provider, nodes)| {
4526 b.get(provider).is_some_and(|other| {
4527 nodes.len() == other.len()
4528 && nodes.iter().all(|(path, node)| {
4529 other
4530 .get(path)
4531 .is_some_and(|o| diff_properties(node, o, path, EXPORTS).is_empty())
4532 })
4533 })
4534 })
4535}
4536
4537const EXPORTS: CanonOpts = CanonOpts {
4541 strip_values: true,
4542 created: false,
4543 exports: true,
4544};
4545
4546const RECREATED: CanonOpts = CanonOpts {
4552 strip_values: true,
4553 created: true,
4554 exports: false,
4555};
4556
4557fn recreated_at(path: &str, tops: &[String]) -> bool {
4560 tops.iter().any(|top| {
4561 path == top
4562 || (path.len() > top.len() + 1
4563 && path.starts_with(top.as_str())
4564 && path.as_bytes()[top.len()] == b'/')
4565 })
4566}
4567
4568async fn settled_reads(
4572 gateway: &dyn ProjectGateway,
4573 names: &[&str],
4574 settle: Duration,
4575 poll: Duration,
4576) -> Result<(ProviderReads, usize, bool, f64), CoreError> {
4577 let started = Instant::now();
4578 let mut reads = 1;
4579 let mut last = retry_transient(|| read_providers(gateway, names)).await?;
4580 let mut settled = false;
4581 loop {
4582 let elapsed = started.elapsed();
4583 if elapsed >= settle {
4584 break;
4585 }
4586 tokio::time::sleep(poll.min(settle - elapsed)).await;
4587 reads += 1;
4588 let next = match read_providers(gateway, names).await {
4591 Ok(next) => next,
4592 Err(err) if read_transient(&err) => continue,
4593 Err(err) => return Err(err),
4594 };
4595 let agree = reads_agree(&last, &next);
4596 last = next;
4597 if agree {
4598 settled = true;
4599 break;
4600 }
4601 }
4602 let elapsed_s = (started.elapsed().as_millis() as f64) / 1000.0;
4603 Ok((last, reads, settled, elapsed_s))
4604}
4605
4606#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
4609#[serde(rename_all = "snake_case")]
4610pub enum RollbackCause {
4611 TagStepFailed,
4614 TagVerifyFailed,
4616 ProjectImportFailed,
4619 ProjectUnverified,
4622 RollbackRun,
4624}
4625
4626impl RollbackCause {
4627 pub fn as_str(self) -> &'static str {
4629 match self {
4630 RollbackCause::TagStepFailed => "tag_step_failed",
4631 RollbackCause::TagVerifyFailed => "tag_verify_failed",
4632 RollbackCause::ProjectImportFailed => "project_import_failed",
4633 RollbackCause::ProjectUnverified => "project_unverified",
4634 RollbackCause::RollbackRun => "rollback_run",
4635 }
4636 }
4637}
4638
4639enum GroupsDone {
4641 Committed {
4643 record: Box<RunRecord>,
4645 providers: Vec<String>,
4647 },
4648 Restored,
4650 Unrestored(String),
4652}
4653
4654#[derive(Default)]
4657struct TagWrites {
4658 plan: Option<TagPlan>,
4661 pre: ProviderReads,
4664 created: BTreeMap<String, CreatedProvider>,
4669 modified: Vec<(String, TagProviderRecord)>,
4672 applied: BTreeMap<String, AppliedSteps>,
4674 groups: Option<GroupsDone>,
4676 sent: Vec<String>,
4679 carried: Vec<String>,
4681 verify: Option<TagVerify>,
4683 host: Option<HostCopy>,
4686 recreate: BTreeMap<String, BTreeSet<String>>,
4690 records_back: Option<Vec<(String, Option<Vec<u8>>)>>,
4694 release_back: Option<Option<Vec<u8>>>,
4698 delete_absent: Vec<String>,
4701}
4702
4703#[derive(Clone, Copy)]
4706struct UndoWriters<'a> {
4707 run: Option<&'a dyn ProjectRunWriter>,
4708 records: Option<&'a dyn TagRecordWriter>,
4709 release: Option<&'a dyn ReleaseRecordWriter>,
4712}
4713
4714struct HostCopy {
4717 run: PreImageRun,
4718 journal: ProjectJournal,
4719}
4720
4721#[derive(Clone, Copy)]
4725struct HostWrites<'a> {
4726 writer: Option<&'a dyn ProjectRunWriter>,
4727 project_zip: Option<&'a [u8]>,
4728}
4729
4730struct PreImageInputs<'a> {
4732 project_zip: Option<&'a [u8]>,
4733 release: Option<&'a [u8]>,
4736 exports: &'a BTreeMap<String, Vec<u8>>,
4738 settings: &'a [(ProviderSettingsPlan, Option<TagProviderRecord>)],
4741 records: &'a BTreeMap<String, TagRecord>,
4743 plan: &'a TagPlan,
4745 groups: bool,
4747}
4748
4749fn action_word(action: ProjectAction) -> &'static str {
4751 match action {
4752 ProjectAction::Create => "create",
4753 ProjectAction::Skip => "skip",
4754 ProjectAction::Import => "import",
4755 ProjectAction::Record => "record",
4756 }
4757}
4758
4759async fn write_host_copy(
4762 writer: &dyn ProjectRunWriter,
4763 plan: &DeployProjectsResult,
4764 loaded: &LoadedProject,
4765 run_id: &RunId,
4766 pre: &PreImageInputs<'_>,
4767 events: &mut RunEvents<'_>,
4768) -> Result<HostCopy, CoreError> {
4769 let gateway = &loaded.target.gateway;
4770 let project = &loaded.file.project;
4771 let mut blobs: Vec<(String, std::borrow::Cow<'_, [u8]>)> = Vec::new();
4772 if let Some(zip) = pre.project_zip {
4773 blobs.push(("project.zip".to_string(), zip.into()));
4774 }
4775 if let Some(release) = pre.release {
4776 blobs.push((PreImageFile::Release.file_name(), release.into()));
4777 }
4778 let owned = sorted(&loaded.owned_tags);
4779 for provider in &owned {
4780 let name = provider.name.as_str();
4781 if let Some(bytes) = pre.exports.get(name) {
4782 blobs.push((format!("tags-{name}.json"), bytes.as_slice().into()));
4783 }
4784 if let Some((_, Some(record))) = pre.settings.iter().find(|(p, _)| p.provider == name) {
4785 let bytes = serde_json::to_vec_pretty(record).map_err(|e| {
4786 CoreError::Internal(format!("tag provider {name}: its record: {e}"))
4787 })?;
4788 blobs.push((format!("provider-{name}.json"), bytes.into()));
4789 }
4790 if let Some(record) = pre.records.get(name) {
4791 blobs.push((format!("record-{name}.json"), record.to_bytes().into()));
4792 }
4793 if pre.exports.contains_key(name)
4797 && let Some(provider_plan) = pre.plan.providers.iter().find(|p| p.provider == name)
4798 && !provider_plan.added.is_empty()
4799 {
4800 let sent: BTreeMap<&str, &Value> = provider_plan
4801 .added
4802 .iter()
4803 .filter_map(|path| {
4804 provider
4805 .nodes
4806 .get(path)
4807 .map(|node| (path.as_str(), &node.json))
4808 })
4809 .collect();
4810 let bytes = serde_json::to_vec(&sent).map_err(|e| {
4811 CoreError::Internal(format!("tag provider {name}: the sent nodes: {e}"))
4812 })?;
4813 blobs.push((format!("sent-{name}.json"), bytes.into()));
4814 }
4815 }
4816 let mut files = BTreeMap::new();
4817 for (name, bytes) in &blobs {
4818 writer
4819 .write_pre_image(gateway, project, run_id.as_str(), name, bytes)
4820 .await?;
4821 files.insert(name.clone(), sha256_hex(bytes));
4822 }
4823 drop(blobs);
4824 let providers = owned
4825 .iter()
4826 .map(|provider| {
4827 let name = provider.name.as_str();
4828 let provider_plan = pre.plan.providers.iter().find(|p| p.provider == name);
4829 let mut added = provider_plan.map(|p| p.added.clone()).unwrap_or_default();
4830 let mut changed: Vec<String> = provider_plan
4831 .map(|p| p.changed.iter().map(|c| c.path.clone()).collect())
4832 .unwrap_or_default();
4833 for list in [&mut added, &mut changed] {
4834 list.sort_by(|a, b| a.as_bytes().cmp(b.as_bytes()));
4835 list.dedup();
4836 }
4837 let mut removed: Vec<String> =
4838 provider_plan.map(|p| p.removed.clone()).unwrap_or_default();
4839 removed.sort_by(|a, b| a.as_bytes().cmp(b.as_bytes()));
4840 removed.dedup();
4841 PreImageProvider {
4842 name: name.to_string(),
4843 settings: pre
4844 .settings
4845 .iter()
4846 .find(|(p, _)| p.provider == name)
4847 .map_or("same", |(p, _)| p.action.as_str())
4848 .to_string(),
4849 added,
4850 changed,
4851 created: None,
4852 removed,
4853 deleted: Vec::new(),
4854 sent: Some(Vec::new()),
4855 collided: Vec::new(),
4856 }
4857 })
4858 .collect();
4859 let run = PreImageRun {
4860 schema_version: 1,
4861 gateway: gateway.clone(),
4862 project: project.clone(),
4863 run_id: run_id.as_str().to_string(),
4864 release: loaded.release.tag.clone(),
4865 written_at: rfc3339_now_utc(),
4866 project_action: action_word(plan.action).to_string(),
4867 groups: pre.groups,
4868 providers,
4869 files,
4870 };
4871 writer
4872 .write_pre_image(
4873 gateway,
4874 project,
4875 run_id.as_str(),
4876 "run.json",
4877 &run.to_bytes(),
4878 )
4879 .await?;
4880 events.emit(ProjectsEvent::HostPreImageWritten {
4881 files: run.files.len() + 1,
4882 })?;
4883 let journal = ProjectJournal {
4884 schema_version: 1,
4885 gateway: gateway.clone(),
4886 project: project.clone(),
4887 run_id: run_id.as_str().to_string(),
4888 state: JournalState::Applying,
4889 started_at: rfc3339_now_utc(),
4890 providers: owned.iter().map(|p| p.name.clone()).collect(),
4891 };
4892 writer
4893 .write_journal(gateway, project, &journal.to_bytes())
4894 .await?;
4895 events.emit(ProjectsEvent::JournalWritten {
4896 state: JournalState::Applying.as_str(),
4897 })?;
4898 Ok(HostCopy { run, journal })
4899}
4900
4901async fn persist_created(
4906 loaded: &LoadedProject,
4907 writer: Option<&dyn ProjectRunWriter>,
4908 writes: &mut TagWrites,
4909 name: &str,
4910) -> Result<(), CoreError> {
4911 let TagWrites { host, created, .. } = writes;
4912 let (Some(writer), Some(host), Some(created)) = (writer, host.as_mut(), created.get(name))
4913 else {
4914 return Ok(());
4915 };
4916 if let Some(entry) = host.run.providers.iter_mut().find(|p| p.name == name) {
4917 entry.created = Some(CreatedProof {
4918 confirmed: created.confirmed,
4919 signature: created.signature.clone(),
4920 enabled: created.sent.0,
4921 description: created.sent.1.clone(),
4922 config: created.sent.2.clone(),
4923 });
4924 }
4925 writer
4926 .write_pre_image(
4927 &loaded.target.gateway,
4928 &loaded.file.project,
4929 &host.run.run_id,
4930 "run.json",
4931 &host.run.to_bytes(),
4932 )
4933 .await
4934}
4935
4936async fn persist_deleted(
4941 host: Option<(&dyn ProjectRunWriter, &LoadedProject)>,
4942 writes: &mut TagWrites,
4943 name: &str,
4944 paths: &[String],
4945) -> Result<(), CoreError> {
4946 let (Some((writer, loaded)), Some(copy)) = (host, writes.host.as_mut()) else {
4947 return Ok(());
4948 };
4949 let Some(entry) = copy.run.providers.iter_mut().find(|p| p.name == name) else {
4950 return Err(CoreError::Internal(format!(
4951 "the host pre-image lists no tag provider {name}"
4952 )));
4953 };
4954 entry.deleted.extend(paths.iter().cloned());
4955 entry.deleted.sort_by(|a, b| a.as_bytes().cmp(b.as_bytes()));
4956 entry.deleted.dedup();
4957 writer
4958 .write_pre_image(
4959 &loaded.target.gateway,
4960 &loaded.file.project,
4961 ©.run.run_id,
4962 "run.json",
4963 ©.run.to_bytes(),
4964 )
4965 .await
4966}
4967
4968async fn persist_sent(
4973 host: Option<(&dyn ProjectRunWriter, &LoadedProject)>,
4974 writes: &mut TagWrites,
4975 name: &str,
4976 paths: &[String],
4977 collided: &[String],
4978) -> Result<(), CoreError> {
4979 let (Some((writer, loaded)), Some(copy)) = (host, writes.host.as_mut()) else {
4980 return Ok(());
4981 };
4982 let Some(entry) = copy.run.providers.iter_mut().find(|p| p.name == name) else {
4983 return Err(CoreError::Internal(format!(
4984 "the host pre-image lists no tag provider {name}"
4985 )));
4986 };
4987 for (list, more) in [
4988 (entry.sent.get_or_insert_with(Vec::new), paths),
4989 (&mut entry.collided, collided),
4990 ] {
4991 list.extend(more.iter().cloned());
4992 list.sort_by(|a, b| a.as_bytes().cmp(b.as_bytes()));
4993 list.dedup();
4994 }
4995 writer
4996 .write_pre_image(
4997 &loaded.target.gateway,
4998 &loaded.file.project,
4999 ©.run.run_id,
5000 "run.json",
5001 ©.run.to_bytes(),
5002 )
5003 .await
5004}
5005
5006fn host_journal_location(loaded: &LoadedProject) -> String {
5008 project_journal_location(
5009 LedgerRoot::for_node(&loaded.target.nodes.master),
5010 &loaded.target.gateway,
5011 &loaded.file.project,
5012 )
5013}
5014
5015async fn clear_host_journal(
5019 result: &mut DeployProjectsResult,
5020 loaded: &LoadedProject,
5021 writer: Option<&dyn ProjectRunWriter>,
5022 host: Option<&HostCopy>,
5023 events: &mut RunEvents<'_>,
5024) {
5025 let (Some(writer), Some(_)) = (writer, host) else {
5026 return;
5027 };
5028 match writer
5029 .clear_journal(&loaded.target.gateway, &loaded.file.project)
5030 .await
5031 {
5032 Ok(()) => events.emit_late(ProjectsEvent::JournalCleared),
5033 Err(err) => result.warnings.push(format!(
5034 "the host journal {} could not be cleared ({}); the next run of this project refuses \
5035 until it is removed",
5036 host_journal_location(loaded),
5037 err.code()
5038 )),
5039 }
5040}
5041
5042async fn unwritten(
5048 err: CoreError,
5049 loaded: &LoadedProject,
5050 writer: Option<&dyn ProjectRunWriter>,
5051 host: Option<&HostCopy>,
5052 events: &mut RunEvents<'_>,
5053) -> CoreError {
5054 if let (Some(writer), Some(_)) = (writer, host)
5055 && writer
5056 .clear_journal(&loaded.target.gateway, &loaded.file.project)
5057 .await
5058 .is_ok()
5059 {
5060 events.emit_late(ProjectsEvent::JournalCleared);
5061 }
5062 err
5063}
5064
5065async fn journal_rolling_back(
5068 result: &mut DeployProjectsResult,
5069 loaded: &LoadedProject,
5070 writer: Option<&dyn ProjectRunWriter>,
5071 host: Option<&mut HostCopy>,
5072 events: &mut RunEvents<'_>,
5073) {
5074 let (Some(writer), Some(host)) = (writer, host) else {
5075 return;
5076 };
5077 if host.journal.state == JournalState::RollingBack {
5078 return;
5080 }
5081 host.journal.state = JournalState::RollingBack;
5082 host.journal.started_at = rfc3339_now_utc();
5083 match writer
5084 .write_journal(
5085 &loaded.target.gateway,
5086 &loaded.file.project,
5087 &host.journal.to_bytes(),
5088 )
5089 .await
5090 {
5091 Ok(()) => events.emit_late(ProjectsEvent::JournalWritten {
5092 state: JournalState::RollingBack.as_str(),
5093 }),
5094 Err(err) => result.warnings.push(format!(
5095 "the host journal {} could not be switched to rolling_back ({}); it still names this \
5096 run",
5097 host_journal_location(loaded),
5098 err.code()
5099 )),
5100 }
5101}
5102
5103impl TagWrites {
5104 fn wrote(&self) -> bool {
5106 !self.sent.is_empty()
5107 }
5108}
5109
5110struct CreatedProvider {
5113 confirmed: bool,
5117 sent: (bool, Option<String>, Value),
5119 signature: Option<String>,
5124}
5125
5126const READ_ATTEMPTS: usize = 3;
5130const READ_PAUSE: std::time::Duration = std::time::Duration::from_secs(2);
5131
5132fn read_transient(err: &CoreError) -> bool {
5135 match err {
5136 CoreError::Internal(_) | CoreError::GatewayRestarting { .. } => true,
5137 CoreError::Network {
5138 source: Some(source),
5139 ..
5140 } => source.is_timeout(),
5141 _ => false,
5142 }
5143}
5144
5145async fn retry_transient<T, F, Fut>(mut read: F) -> Result<T, CoreError>
5151where
5152 F: FnMut() -> Fut,
5153 Fut: std::future::Future<Output = Result<T, CoreError>>,
5154{
5155 let mut attempt = 1;
5156 loop {
5157 match read().await {
5158 Err(err) if attempt < READ_ATTEMPTS && read_transient(&err) => {
5159 attempt += 1;
5160 tokio::time::sleep(READ_PAUSE).await;
5161 }
5162 answer => return answer,
5163 }
5164 }
5165}
5166
5167fn holds_sent(record: &TagProviderRecord, sent: &(bool, Option<String>, Value)) -> bool {
5170 record.enabled == sent.0
5171 && record.description() == sent.1.as_deref()
5172 && canonical(&record.config) == canonical(&sent.2)
5173}
5174
5175struct TagFailure {
5179 cause: RollbackCause,
5180 what: String,
5181 writes: TagWrites,
5182}
5183
5184enum TagStepError {
5186 Unwritten(CoreError),
5188 Written(Box<TagFailure>),
5190}
5191
5192impl From<CoreError> for TagStepError {
5193 fn from(err: CoreError) -> Self {
5194 TagStepError::Unwritten(err)
5195 }
5196}
5197
5198enum Stop {
5200 Error(CoreError),
5203 Failed(RollbackCause, String),
5205}
5206
5207impl From<CoreError> for Stop {
5208 fn from(err: CoreError) -> Self {
5209 Stop::Error(err)
5210 }
5211}
5212
5213#[allow(clippy::too_many_arguments)]
5224async fn deploy_owned_tags(
5225 plan: &DeployProjectsResult,
5226 tag_records: &BTreeMap<String, TagRecord>,
5227 loaded: &LoadedProject,
5228 input: &ProjectsInput,
5229 surfaces: &ProjectsSurfaces<'_>,
5230 opts: &ProjectsRunOptions<'_>,
5231 run_dir: &RunDir,
5232 run_id: &RunId,
5233 host: HostWrites<'_>,
5234 route: &RouteState<'_>,
5235 events: &mut RunEvents<'_>,
5236) -> Result<TagStep, TagStepError> {
5237 let gateway = surfaces.gateway;
5238 let Some(report) = &plan.tags else {
5239 return Ok(TagStep {
5240 carried: Vec::new(),
5241 verify: None,
5242 batches_sent: 0,
5243 other_writes: 0,
5244 writes: TagWrites::default(),
5245 deleted: Vec::new(),
5246 absent: Vec::new(),
5247 });
5248 };
5249 let settings = fresh_provider_settings(loaded, gateway, report, run_dir, run_id).await?;
5253 let creating: BTreeSet<&str> = report
5254 .provider_settings
5255 .iter()
5256 .filter(|p| p.action == ProviderSettingsAction::Create)
5257 .map(|p| p.provider.as_str())
5258 .collect();
5259 let mut live = BTreeMap::new();
5263 let mut raw: BTreeMap<String, Vec<u8>> = BTreeMap::new();
5264 for provider in &loaded.owned_tags {
5265 if creating.contains(provider.name.as_str()) {
5266 continue;
5267 }
5268 let bytes = gateway.tag_export(&provider.name).await?;
5269 let mut file = run_dir
5270 .create_private_file_nested(&["export", "tags"], &format!("{}.json", provider.name))?;
5271 file.write_all(&bytes)
5272 .and_then(|()| file.sync_all())
5273 .map_err(|e| CoreError::DeployRunInvalid {
5274 run_id: run_id.as_str().to_string(),
5275 problem: format!(
5276 "could not write the tag pre-image of provider {} ({:?})",
5277 provider.name,
5278 e.kind()
5279 ),
5280 })?;
5281 events.emit(ProjectsEvent::TagsPreImageExported {
5282 provider: provider.name.clone(),
5283 bytes: bytes.len(),
5284 })?;
5285 live.insert(
5286 provider.name.clone(),
5287 flatten_export(&bytes, &provider.name)?,
5288 );
5289 raw.insert(provider.name.clone(), bytes);
5290 }
5291 let now = plan_tags(&loaded.owned_tags, &live, tag_records);
5293 if now.live_sha256 != report.live_sha256 {
5294 return Err(CoreError::PlanStale {
5295 expected: report.live_sha256.clone(),
5296 actual: now.live_sha256,
5297 detail: "the owned tag providers changed since the plan".to_string(),
5298 }
5299 .into());
5300 }
5301 let groups = fresh_groups(loaded, input, surfaces, report, &now, tag_records, &live).await?;
5304 let empty = BTreeMap::new();
5307 let owned = sorted(&loaded.owned_tags);
5308 let mut queue: VecDeque<ImportBatch> = VecDeque::new();
5309 let mut over_cap = Vec::new();
5310 for (provider_plan, provider) in now.providers.iter().zip(&owned) {
5311 let nodes = live.get(&provider.name).unwrap_or(&empty);
5312 queue.extend(import_batches(
5313 provider_plan,
5314 provider,
5315 nodes,
5316 REMOVED_PROPERTY_POLICY,
5317 )?);
5318 over_cap.extend(rollback_over_cap(provider_plan, provider, nodes));
5321 }
5322 if !over_cap.is_empty() {
5323 return Err(CoreError::deploy_findings(over_cap).into());
5324 }
5325 let mut refused = Vec::new();
5329 for batch in queue.iter().filter(|batch| batch.carry) {
5330 let nodes = live.get(&batch.provider).unwrap_or(&empty);
5331 for (path, found) in uncarriable_in_batch(batch, nodes)? {
5332 let file = owned
5333 .iter()
5334 .find(|p| p.name == batch.provider)
5335 .and_then(|p| p.nodes.get(&path))
5336 .map_or_else(
5337 || full_tag_path(&batch.provider, &path),
5338 |node| node.file.clone(),
5339 );
5340 refused.push(DeployFinding {
5341 file,
5342 line: None,
5343 column: None,
5344 message: format!(
5345 "{} hold(s) a live value and a valueSource that names no source (not a \
5346 string): the repo drops a property, so the tag would go Overwrite, which \
5347 cannot carry that value and would reset it; nothing was written",
5348 found
5349 .iter()
5350 .map(|p| full_tag_path(&batch.provider, p))
5351 .collect::<Vec<_>>()
5352 .join(", ")
5353 ),
5354 });
5355 }
5356 }
5357 if !refused.is_empty() {
5358 return Err(CoreError::deploy_findings(refused).into());
5359 }
5360
5361 let mut writes = TagWrites::default();
5365 if let Some(writer) = host.writer {
5366 let release = surfaces
5367 .host
5368 .read_release_record(&loaded.target.gateway, &loaded.file.project)
5369 .await?;
5370 let pre = PreImageInputs {
5371 project_zip: host.project_zip,
5372 release: release.as_deref(),
5373 exports: &raw,
5374 settings: &settings,
5375 records: tag_records,
5376 plan: &now,
5377 groups: groups.as_ref().is_some_and(GroupWork::would_change),
5378 };
5379 writes.host = Some(write_host_copy(writer, plan, loaded, run_id, &pre, events).await?);
5380 }
5381 drop(raw);
5382 let outcome = send_owned_tags(
5383 loaded,
5384 surfaces,
5385 opts,
5386 run_id,
5387 report,
5388 &settings,
5389 groups.as_ref(),
5390 &now,
5391 &live,
5392 queue,
5393 &mut writes,
5394 host.writer,
5395 route,
5396 events,
5397 )
5398 .await;
5399 writes.plan = Some(now);
5400 writes.pre = live;
5401 match outcome {
5402 Ok((verify, other_writes)) => {
5403 let mut carried = writes.carried.clone();
5404 carried.sort_by(|a, b| a.as_bytes().cmp(b.as_bytes()));
5405 let mut deleted: Vec<String> = writes
5406 .applied
5407 .iter()
5408 .flat_map(|(name, applied)| applied.deleted.iter().map(|p| full_tag_path(name, p)))
5409 .collect();
5410 deleted.sort_by(|a, b| a.as_bytes().cmp(b.as_bytes()));
5411 let absent = writes.delete_absent.clone();
5412 Ok(TagStep {
5413 carried,
5414 verify: Some(verify),
5415 batches_sent: writes.sent.len().saturating_sub(other_writes),
5416 other_writes,
5417 writes,
5418 deleted,
5419 absent,
5420 })
5421 }
5422 Err(Stop::Error(err)) if !writes.wrote() => {
5423 Err(TagStepError::Unwritten(
5425 unwritten(err, loaded, host.writer, writes.host.as_ref(), events).await,
5426 ))
5427 }
5428 Err(Stop::Error(err)) => Err(TagStepError::Written(Box::new(TagFailure {
5429 cause: RollbackCause::TagStepFailed,
5430 what: format!("the tag step stopped ({})", err.code()),
5431 writes,
5432 }))),
5433 Err(Stop::Failed(cause, what)) => Err(TagStepError::Written(Box::new(TagFailure {
5434 cause,
5435 what,
5436 writes,
5437 }))),
5438 }
5439}
5440
5441#[allow(clippy::too_many_arguments)]
5448async fn send_owned_tags(
5449 loaded: &LoadedProject,
5450 surfaces: &ProjectsSurfaces<'_>,
5451 opts: &ProjectsRunOptions<'_>,
5452 run_id: &RunId,
5453 report: &TagsReport,
5454 settings: &[(ProviderSettingsPlan, Option<TagProviderRecord>)],
5455 groups: Option<&GroupWork>,
5456 now: &TagPlan,
5457 live: &ProviderReads,
5458 mut queue: VecDeque<ImportBatch>,
5459 writes: &mut TagWrites,
5460 run_writer: Option<&dyn ProjectRunWriter>,
5461 route: &RouteState<'_>,
5462 events: &mut RunEvents<'_>,
5463) -> Result<(TagVerify, usize), Stop> {
5464 let gateway = surfaces.gateway;
5465 let failed = RollbackCause::TagStepFailed;
5466 let repos = settings
5469 .iter()
5470 .map(|(planned, _)| provider_settings_of(loaded, &planned.provider))
5471 .collect::<Result<Vec<_>, _>>()?;
5472 let mut other_writes = 0;
5473 for ((planned, record), repo) in settings.iter().zip(repos) {
5474 if planned.action == ProviderSettingsAction::Same {
5475 continue;
5476 }
5477 let name = planned.provider.as_str();
5478 let answer = match (planned.action, record) {
5479 (ProviderSettingsAction::Create, _) => {
5480 let sent = (repo.enabled, repo.description.clone(), repo.config.clone());
5481 let answer = gateway
5482 .tag_provider_create(&[TagProviderCreate::with_settings(
5483 name,
5484 repo.enabled,
5485 repo.description.clone(),
5486 repo.config.clone(),
5487 )])
5488 .await;
5489 match &answer {
5495 Ok(()) => {
5496 writes.created.insert(
5497 name.to_string(),
5498 CreatedProvider {
5499 confirmed: true,
5500 sent,
5501 signature: None,
5502 },
5503 );
5504 let _ = persist_created(loaded, run_writer, writes, name).await;
5511 }
5512 Err(err) if import_outcome_unknown(err) => {
5513 let signature =
5514 match retry_transient(|| gateway.tag_provider_record(name)).await {
5515 Ok(Some(record)) if holds_sent(&record, &sent) => record.signature,
5516 _ => None,
5517 };
5518 writes.created.insert(
5519 name.to_string(),
5520 CreatedProvider {
5521 confirmed: false,
5522 sent,
5523 signature,
5524 },
5525 );
5526 let _ = persist_created(loaded, run_writer, writes, name).await;
5530 }
5531 Err(_) => {}
5532 }
5533 answer
5534 }
5535 (ProviderSettingsAction::Modify, Some(record)) => {
5536 writes.modified.push((name.to_string(), record.clone()));
5537 let signature = record.signature.clone().unwrap_or_default();
5538 gateway
5539 .tag_provider_modify(&[TagProviderModify {
5540 name: name.to_string(),
5541 collection: record.collection().unwrap_or("core").to_string(),
5542 enabled: repo.enabled,
5543 description: repo.description.clone(),
5544 config: repo.config.clone(),
5545 signature,
5546 }])
5547 .await
5548 }
5549 _ => {
5550 return Err(Stop::Error(CoreError::Internal(format!(
5551 "tag provider {name} is planned to be modified but has no record"
5552 ))));
5553 }
5554 };
5555 writes.sent.push(format!(
5557 "[{name}] provider settings ({})",
5558 planned.action.as_str()
5559 ));
5560 other_writes += 1;
5561 if let Err(err) = answer {
5562 return Err(Stop::Failed(
5563 failed,
5564 format!(
5565 "the settings of tag provider {name} were not applied ({})",
5566 err.code()
5567 ),
5568 ));
5569 }
5570 let reread = retry_transient(|| gateway.tag_provider_record(name))
5574 .await
5575 .map_err(|err| {
5576 Stop::Failed(
5577 failed,
5578 format!(
5579 "tag provider {name} could not be read after its settings were applied ({})",
5580 err.code()
5581 ),
5582 )
5583 })?;
5584 if let Some(created) = writes.created.get_mut(name) {
5587 created.signature = reread.as_ref().and_then(|r| r.signature.clone());
5588 if let Err(err) = persist_created(loaded, run_writer, writes, name).await {
5591 return Err(Stop::Failed(
5592 failed,
5593 format!(
5594 "the host pre-image could not record tag provider {name}, which this run \
5595 created ({})",
5596 err.code()
5597 ),
5598 ));
5599 }
5600 }
5601 let check = plan_provider_settings(name, repo, reread.as_ref());
5602 if check.action != ProviderSettingsAction::Same {
5603 return Err(Stop::Failed(
5604 failed,
5605 format!(
5606 "the settings of tag provider {name} read back different from the repo ({})",
5607 if check.properties.is_empty() {
5608 "the provider is absent".to_string()
5609 } else {
5610 check.properties.join(", ")
5611 }
5612 ),
5613 ));
5614 }
5615 events.emit_late(ProjectsEvent::TagProviderSettingsApplied {
5616 provider: name.to_string(),
5617 action: planned.action.as_str(),
5618 });
5619 }
5620
5621 if let Some(work) = groups.filter(|work| work.would_change()) {
5624 let providers = work
5625 .plans
5626 .iter()
5627 .filter(|p| p.would_change())
5628 .map(|p| p.provider.clone())
5629 .collect::<Vec<_>>();
5630 let label = format!("tag groups of {}", providers.join(", "));
5631 match commit_groups(loaded, surfaces, run_id, &report.plan_sha256, work).await {
5632 Ok((written, deleted, record)) => {
5633 writes
5634 .sent
5635 .push(format!("{label} ({written} written, {deleted} deleted)"));
5636 other_writes += 1;
5637 writes.groups = Some(GroupsDone::Committed {
5638 record: Box::new(record),
5639 providers: providers.clone(),
5640 });
5641 events.emit_late(ProjectsEvent::TagGroupsWritten {
5642 providers,
5643 written,
5644 deleted,
5645 });
5646 }
5647 Err(GroupFailure::NotWritten(err)) if !writes.wrote() => return Err(err.into()),
5648 Err(GroupFailure::NotWritten(err)) => {
5649 return Err(Stop::Failed(
5650 failed,
5651 format!("the {label} were not written ({})", err.code()),
5652 ));
5653 }
5654 Err(GroupFailure::Stale) if !writes.wrote() => {
5655 return Err(CoreError::PlanStale {
5656 expected: "the planned tag-group units".to_string(),
5657 actual: "a unit changed before the commit".to_string(),
5658 detail: "the owned providers' tag-group units changed since the plan"
5659 .to_string(),
5660 }
5661 .into());
5662 }
5663 Err(GroupFailure::Stale) => {
5664 return Err(Stop::Failed(
5665 failed,
5666 format!("the {label} were not written: a unit changed since the plan"),
5667 ));
5668 }
5669 Err(GroupFailure::Restored(why)) => {
5670 writes.sent.push(format!("{label} (restored)"));
5671 writes.groups = Some(GroupsDone::Restored);
5672 return Err(Stop::Failed(
5673 failed,
5674 format!(
5675 "the {label} did not verify ({why}); the group units were restored from \
5676 the run's snapshot"
5677 ),
5678 ));
5679 }
5680 Err(GroupFailure::Unrestored(why)) => {
5681 writes.sent.push(format!("{label} (not restored)"));
5682 let journal = journal_location(
5683 LedgerRoot::for_node(&loaded.target.nodes.master),
5684 &loaded.target.gateway,
5685 &loaded.target.mode,
5686 );
5687 writes.groups = Some(GroupsDone::Unrestored(format!(
5688 "the {label} failed and could not be restored ({why}); the on-box journal \
5689 {journal} stays"
5690 )));
5691 return Err(Stop::Failed(
5692 failed,
5693 format!("the {label} failed and could not be restored ({why})"),
5694 ));
5695 }
5696 }
5697 }
5698
5699 while let Some(batch) = queue.pop_front() {
5700 let name = batch.provider.clone();
5701 let body = if batch.carry {
5702 let fresh = match read_flat(gateway, &name).await {
5704 Ok(fresh) => fresh,
5705 Err(err) if !writes.wrote() => return Err(err.into()),
5706 Err(err) => {
5707 return Err(Stop::Failed(
5708 failed,
5709 format!(
5710 "provider {name} could not be read to carry the live values ({})",
5711 err.code()
5712 ),
5713 ));
5714 }
5715 };
5716 let refused = uncarriable_in_batch(&batch, &fresh)?;
5719 if !refused.is_empty() {
5720 let paths: Vec<String> = refused
5721 .iter()
5722 .flat_map(|(_, found)| found.iter().map(|p| full_tag_path(&name, p)))
5723 .collect();
5724 let what = format!(
5725 "{} hold(s) a live value and a valueSource that names no source in the \
5726 export read right before their Overwrite, which would reset it, so the \
5727 batch was not sent",
5728 paths.join(", ")
5729 );
5730 if !writes.wrote() {
5731 return Err(CoreError::deploy_tree_invalid(
5732 full_tag_path(&name, &batch.path),
5733 None,
5734 None,
5735 what,
5736 )
5737 .into());
5738 }
5739 return Err(Stop::Failed(failed, what));
5740 }
5741 match carried(&batch, &fresh)? {
5742 Carried::Body(bytes) => bytes,
5743 Carried::Split(parts) => {
5744 for part in parts.into_iter().rev() {
5745 queue.push_front(part);
5746 }
5747 continue;
5748 }
5749 }
5750 } else {
5751 batch.body.clone()
5752 };
5753 if batch.carry {
5754 let event = ProjectsEvent::TagValuesCarried {
5755 provider: name.clone(),
5756 path: batch.path.clone(),
5757 nodes: batch.nodes.len(),
5758 };
5759 if writes.wrote() {
5760 events.emit_late(event);
5761 } else {
5762 events.emit(event)?;
5763 }
5764 writes
5765 .carried
5766 .extend(batch.nodes.iter().map(|p| full_tag_path(&name, p)));
5767 }
5768 let event = ProjectsEvent::TagImportSent {
5769 provider: name.clone(),
5770 path: batch.path.clone(),
5771 policy: batch.policy.as_wire(),
5772 nodes: batch.nodes.len(),
5773 carry: batch.carry,
5774 };
5775 if writes.wrote() {
5776 events.emit_late(event);
5777 } else {
5778 events.emit(event)?;
5779 }
5780 let host = run_writer.map(|writer| (writer, loaded));
5784 if let Err(err) = persist_sent(host, writes, &name, &batch.nodes, &[]).await {
5785 if !writes.wrote() {
5786 return Err(Stop::Error(err));
5787 }
5788 return Err(Stop::Failed(
5789 failed,
5790 format!(
5791 "the host pre-image could not record the tag import into provider {name} \
5792 ({}), so it was not sent",
5793 err.code()
5794 ),
5795 ));
5796 }
5797 let answer = gateway
5798 .tag_import(&name, &batch.path, batch.policy, body)
5799 .await;
5800 let sent_nodes = sent_bodies(loaded, now, &name, &batch);
5805 let applied = writes.applied.entry(name.clone()).or_default();
5806 applied.sent(&batch);
5807 applied.bodies.extend(sent_nodes);
5808 writes.sent.push(format!(
5809 "{} ({}, {} node(s))",
5810 full_tag_path(&name, &batch.path),
5811 batch.policy.as_wire(),
5812 batch.nodes.len()
5813 ));
5814 let answer = answer.map_err(|err| {
5815 Stop::Failed(
5816 failed,
5817 format!(
5818 "the tag import into provider {name} failed ({})",
5819 err.code()
5820 ),
5821 )
5822 })?;
5823 let clean = answer.is_clean();
5824 events.emit_late(ProjectsEvent::TagImportAnswered {
5825 provider: name.clone(),
5826 clean,
5827 });
5828 let collided = collided_paths(&name, &batch, &answer);
5832 if !collided.is_empty() {
5833 if let Some(applied) = writes.applied.get_mut(&name) {
5834 applied.collided.extend(collided.iter().cloned());
5835 }
5836 let _ = persist_sent(host, writes, &name, &[], &collided).await;
5839 }
5840 if !clean {
5841 let mut codes = answer.bad_quality.clone();
5842 codes.truncate(TAG_FAILURE_PATHS_SHOWN);
5843 let first = batch.nodes.first().map_or_else(
5844 || full_tag_path(&name, &batch.path),
5845 |p| full_tag_path(&name, p),
5846 );
5847 return Err(Stop::Failed(
5848 failed,
5849 format!(
5850 "the tag import headed by {first} into provider {name} was not clean ({} failure(s){})",
5851 answer.failure_count.unwrap_or(codes.len() as u64),
5852 if codes.is_empty() {
5853 String::new()
5854 } else {
5855 format!(": {}", codes.join(", "))
5856 }
5857 ),
5858 ));
5859 }
5860 }
5861
5862 let mut absent = Vec::new();
5865 let host = run_writer.map(|writer| (writer, loaded));
5866 let sent = delete_removed(gateway, now, route, writes, host, &mut absent, events).await;
5867 writes.delete_absent = absent;
5868 other_writes += sent?;
5869
5870 let owned = sorted(&loaded.owned_tags);
5873 let names = owned_names(loaded);
5874 let wrote = writes.wrote();
5875 let (last, reads, settled, elapsed_s) = settled_reads(
5876 gateway,
5877 &names,
5878 opts.tag_verify_settle,
5879 opts.tag_verify_poll,
5880 )
5881 .await
5882 .map_err(|err| {
5883 if wrote {
5884 Stop::Failed(
5885 RollbackCause::TagVerifyFailed,
5886 format!(
5887 "the owned tag providers could not be read back to verify ({})",
5888 err.code()
5889 ),
5890 )
5891 } else {
5892 Stop::Error(err)
5893 }
5894 })?;
5895
5896 let empty = BTreeMap::new();
5897 let mut mismatches = Vec::new();
5898 for (provider_plan, provider) in now.providers.iter().zip(&owned) {
5899 let name = provider.name.as_str();
5900 let after = last.get(name).unwrap_or(&empty);
5901 let pre = live.get(name).unwrap_or(&empty);
5902 let created: BTreeSet<&str> = provider_plan.added.iter().map(String::as_str).collect();
5903 let mut differ = Vec::new();
5904 for (path, node) in &provider.nodes {
5905 match after.get(path) {
5906 None => differ.push(format!("{} (absent)", full_tag_path(name, path))),
5907 Some(live_node) => {
5908 let properties = diff_properties(
5909 &node.json,
5910 live_node,
5911 path,
5912 CanonOpts {
5913 strip_values: true,
5914 created: created.contains(path.as_str()),
5915 exports: false,
5916 },
5917 );
5918 if !properties.is_empty() {
5919 differ.push(format!(
5920 "{} ({})",
5921 full_tag_path(name, path),
5922 properties.join(", ")
5923 ));
5924 }
5925 }
5926 }
5927 }
5928 for path in &provider_plan.removed {
5930 if after.contains_key(path) {
5931 differ.push(format!("{} (present)", full_tag_path(name, path)));
5932 }
5933 }
5934 for path in &provider_plan.drift {
5936 let properties = match (pre.get(path), after.get(path)) {
5937 (Some(before), Some(now)) => diff_properties(before, now, path, EXPORTS),
5938 (Some(_), None) => vec!["absent".to_string()],
5939 _ => Vec::new(),
5940 };
5941 if !properties.is_empty() {
5942 differ.push(format!(
5943 "{} (drift: {})",
5944 full_tag_path(name, path),
5945 properties.join(", ")
5946 ));
5947 }
5948 }
5949 events.emit_late(ProjectsEvent::TagsVerified {
5950 provider: name.to_string(),
5951 equal: differ.is_empty(),
5952 });
5953 mismatches.extend(differ);
5954 }
5955 mismatches.sort_by(|a, b| a.as_bytes().cmp(b.as_bytes()));
5956 let mismatches_count = mismatches.len();
5957 let what = (mismatches_count > 0).then(|| {
5958 let shown: Vec<&str> = mismatches
5959 .iter()
5960 .take(TAG_FAILURE_PATHS_SHOWN)
5961 .map(String::as_str)
5962 .collect();
5963 format!(
5964 "the tag verify found {mismatches_count} node(s) unlike the repo or their pre-image \
5965 after {reads} read(s){}: {}",
5966 if settled { "" } else { ", unsettled" },
5967 shown.join("; ")
5968 )
5969 });
5970 let mut capped = mismatches;
5971 capped.truncate(LIST_CAP);
5972 let verify = TagVerify {
5973 equal: mismatches_count == 0,
5974 settled,
5975 reads,
5976 elapsed_s,
5977 mismatches: capped,
5978 mismatches_count,
5979 };
5980 writes.verify = Some(verify.clone());
5981 if let Some(what) = what {
5982 if !wrote {
5983 return Err(CoreError::PlanStale {
5985 expected: "the owned tags as planned".to_string(),
5986 actual: format!("{mismatches_count} node(s) unlike the repo"),
5987 detail: what,
5988 }
5989 .into());
5990 }
5991 return Err(Stop::Failed(RollbackCause::TagVerifyFailed, what));
5992 }
5993 Ok((verify, other_writes))
5994}
5995
5996fn sent_bodies(
6000 loaded: &LoadedProject,
6001 now: &TagPlan,
6002 provider: &str,
6003 batch: &ImportBatch,
6004) -> Vec<(String, Value)> {
6005 let Some(plan) = now.providers.iter().find(|p| p.provider == provider) else {
6006 return Vec::new();
6007 };
6008 let Some(repo) = loaded.owned_tags.iter().find(|p| p.name == provider) else {
6009 return Vec::new();
6010 };
6011 batch
6012 .nodes
6013 .iter()
6014 .filter(|path| plan.added.binary_search(path).is_ok())
6016 .filter_map(|path| {
6017 repo.nodes
6018 .get(path)
6019 .map(|node| (path.clone(), node.json.clone()))
6020 })
6021 .collect()
6022}
6023
6024fn collided_paths(provider: &str, batch: &ImportBatch, answer: &TagImportAnswer) -> Vec<String> {
6029 let prefix = format!("[{provider}]");
6030 let mut out = Vec::new();
6031 let mut whole = answer.collisions_unnamed > 0;
6032 for full in &answer.collided {
6033 match full.strip_prefix(&prefix) {
6034 Some(path) if batch.nodes.iter().any(|n| n == path) => out.push(path.to_string()),
6035 _ => whole = true,
6036 }
6037 }
6038 if whole {
6039 return batch.nodes.clone();
6040 }
6041 out.sort_by(|a, b| a.as_bytes().cmp(b.as_bytes()));
6042 out.dedup();
6043 out
6044}
6045
6046async fn delete_removed(
6059 gateway: &dyn ProjectGateway,
6060 now: &TagPlan,
6061 route: &RouteState<'_>,
6062 writes: &mut TagWrites,
6063 host: Option<(&dyn ProjectRunWriter, &LoadedProject)>,
6064 absent: &mut Vec<String>,
6065 events: &mut RunEvents<'_>,
6066) -> Result<usize, Stop> {
6067 let failed = RollbackCause::TagStepFailed;
6068 let mut calls = 0;
6069 for provider_plan in now.providers.iter().filter(|p| !p.removed.is_empty()) {
6070 let name = provider_plan.provider.as_str();
6071 let Some((client, handshake)) = route.ready() else {
6072 return Err(Stop::Error(CoreError::Internal(format!(
6074 "tag provider {name} has removals and the run has no ready tag-delete route"
6075 ))));
6076 };
6077 let fresh = match read_flat(gateway, name).await {
6078 Ok(fresh) => fresh,
6079 Err(err) if !writes.wrote() => return Err(err.into()),
6080 Err(err) => {
6081 return Err(Stop::Failed(
6082 failed,
6083 format!(
6084 "provider {name} could not be read before its deletes ({})",
6085 err.code()
6086 ),
6087 ));
6088 }
6089 };
6090 absent.extend(
6092 provider_plan
6093 .removed
6094 .iter()
6095 .filter(|path| !fresh.contains_key(*path))
6096 .map(|path| full_tag_path(name, path)),
6097 );
6098 let calls_of = match delete_calls(provider_plan, &fresh, handshake, TAG_DELETE_DESCENDANTS)
6099 {
6100 Ok(calls_of) => calls_of,
6101 Err(refused) => {
6102 let what = refused.describe(name);
6103 if !writes.wrote() {
6104 return Err(Stop::Error(CoreError::PlanStale {
6107 expected: "the owned tags as planned".to_string(),
6108 actual: "a removal no longer holds".to_string(),
6109 detail: what,
6110 }));
6111 }
6112 return Err(Stop::Failed(failed, what));
6113 }
6114 };
6115 for chunk in calls_of.into_iter().map(|call| call.paths) {
6116 if let Err(err) = persist_deleted(host, writes, name, &chunk).await {
6120 let what = format!(
6121 "the host pre-image could not record the deletes in provider {name} ({}), so \
6122 they were not sent",
6123 err.code()
6124 );
6125 if !writes.wrote() {
6126 return Err(Stop::Error(err));
6127 }
6128 return Err(Stop::Failed(failed, what));
6129 }
6130 writes
6133 .applied
6134 .entry(name.to_string())
6135 .or_default()
6136 .deleted
6137 .extend(chunk.iter().cloned());
6138 writes
6139 .sent
6140 .push(format!("[{name}] tag delete ({} path(s))", chunk.len()));
6141 calls += 1;
6142 events.emit_late(ProjectsEvent::TagDeleteSent {
6143 provider: name.to_string(),
6144 paths: chunk.clone(),
6145 rollback: false,
6146 });
6147 let answer = match client.delete(name, &chunk).await {
6148 Ok(answer) => answer.check(name, &chunk).map(|()| answer),
6149 Err(err) => Err(err),
6150 };
6151 let answer = answer.map_err(|err| {
6152 Stop::Failed(
6153 failed,
6154 format!("the tag delete in provider {name} failed ({})", err.code()),
6155 )
6156 })?;
6157 let clean = answer.is_clean();
6158 let removed = answer.results.iter().filter(|r| r.removed()).count();
6159 events.emit_late(ProjectsEvent::TagDeleteAnswered {
6160 provider: name.to_string(),
6161 clean,
6162 deleted: removed,
6163 });
6164 if !clean {
6165 let kept: Vec<String> = answer
6166 .results
6167 .iter()
6168 .filter(|r| !r.removed())
6169 .take(TAG_FAILURE_PATHS_SHOWN)
6170 .map(|r| {
6171 format!(
6172 "{} ({})",
6173 full_tag_path(name, &r.path),
6174 r.not_removed_detail()
6175 )
6176 })
6177 .collect();
6178 return Err(Stop::Failed(
6179 failed,
6180 format!("the tag-delete route did not remove {}", kept.join(", ")),
6181 ));
6182 }
6183 }
6184 }
6185 Ok(calls)
6186}
6187
6188struct ProjectUndo {
6191 pre: Option<Vec<u8>>,
6192}
6193
6194async fn project_sha(gateway: &dyn ProjectGateway, project: &str) -> Option<String> {
6197 let zip = retry_transient(|| gateway.export_bytes(project))
6198 .await
6199 .ok()?;
6200 live_digest(project, &zip).ok().map(|digest| digest.sha256)
6201}
6202
6203fn settings_equal(a: &TagProviderRecord, b: &TagProviderRecord) -> bool {
6207 a.enabled == b.enabled
6208 && a.description() == b.description()
6209 && canonical(&a.config) == canonical(&b.config)
6210}
6211
6212async fn restore_provider_settings(
6217 gateway: &dyn ProjectGateway,
6218 name: &str,
6219 pre: &TagProviderRecord,
6220) -> Result<bool, String> {
6221 let now = retry_transient(|| gateway.tag_provider_record(name))
6222 .await
6223 .map_err(|e| format!("it could not be read ({})", e.code()))?
6224 .ok_or_else(|| "it is no longer listed".to_string())?;
6225 if settings_equal(&now, pre) {
6226 return Ok(false);
6227 }
6228 gateway
6229 .tag_provider_modify(&[TagProviderModify {
6230 name: name.to_string(),
6231 collection: pre
6232 .collection()
6233 .or(now.collection())
6234 .unwrap_or("core")
6235 .to_string(),
6236 enabled: pre.enabled,
6237 description: pre.description().map(str::to_string),
6238 config: pre.config.clone(),
6239 signature: now.signature.clone().unwrap_or_default(),
6240 }])
6241 .await
6242 .map_err(|e| format!("the modify back failed ({})", e.code()))?;
6243 match retry_transient(|| gateway.tag_provider_record(name)).await {
6244 Ok(Some(back)) if settings_equal(&back, pre) => Ok(true),
6245 Ok(_) => Err("it does not read back as its pre-image".to_string()),
6246 Err(e) => Err(format!("it could not be read back ({})", e.code())),
6247 }
6248}
6249
6250async fn delete_created_provider(
6259 gateway: &dyn ProjectGateway,
6260 name: &str,
6261 created: &CreatedProvider,
6262) -> Result<bool, String> {
6263 let Some(record) = retry_transient(|| gateway.tag_provider_record(name))
6264 .await
6265 .map_err(|e| format!("it could not be read ({})", e.code()))?
6266 else {
6267 return Ok(false);
6268 };
6269 let Some(signature) = created.signature.as_deref() else {
6270 return Err(if created.confirmed {
6271 "its record could not be read right after the create, so nothing proves it is this \
6272 run's; it was left in place"
6273 .to_string()
6274 } else {
6275 "the create's answer was lost and the provider read right after it did not hold \
6276 what this run sent, so it may be another client's; it was left in place"
6277 .to_string()
6278 });
6279 };
6280 if record.signature.as_deref() != Some(signature) {
6281 return Err(
6282 "it changed since this run's create (its signature is not the one read right after \
6283 the create), so it was left in place"
6284 .to_string(),
6285 );
6286 }
6287 if !created.confirmed && !holds_sent(&record, &created.sent) {
6288 return Err(
6289 "the create's answer was lost and the provider no longer holds what this run sent, \
6290 so it was left in place"
6291 .to_string(),
6292 );
6293 }
6294 gateway
6295 .tag_provider_delete(name, signature)
6296 .await
6297 .map_err(|e| format!("its delete failed ({})", e.code()))?;
6298 match retry_transient(|| gateway.tag_provider_record(name)).await {
6299 Ok(None) => Ok(true),
6300 Ok(Some(_)) => Err("it is still listed after the delete".to_string()),
6301 Err(e) => Err(format!("its delete could not be read back ({})", e.code())),
6302 }
6303}
6304
6305async fn delete_created(
6315 route: &dyn TagDeleteRoute,
6316 handshake: &RouteHandshake,
6317 provider: &str,
6318 paths: &[String],
6319 events: &mut RunEvents<'_>,
6320) -> (Vec<String>, Vec<(String, String)>) {
6321 let mut removed = Vec::new();
6322 let mut kept = Vec::new();
6323 let chunks = match delete_chunks(paths, handshake) {
6324 Ok(chunks) => chunks,
6325 Err(err) => {
6326 let why = format!(
6327 "created by this run; it was not sent to the tag-delete route ({})",
6328 err.code()
6329 );
6330 kept.extend(
6331 paths
6332 .iter()
6333 .map(|p| (full_tag_path(provider, p), why.clone())),
6334 );
6335 return (removed, kept);
6336 }
6337 };
6338 for chunk in chunks.into_iter().filter(|chunk| !chunk.is_empty()) {
6339 events.emit_late(ProjectsEvent::TagDeleteSent {
6340 provider: provider.to_string(),
6341 paths: chunk.clone(),
6342 rollback: true,
6343 });
6344 let answer = match route.delete(provider, &chunk).await {
6345 Ok(answer) => answer.check(provider, &chunk).map(|()| answer),
6346 Err(err) => Err(err),
6347 };
6348 match answer {
6349 Ok(answer) => {
6350 let mut count = 0;
6351 for result in &answer.results {
6352 let full = full_tag_path(provider, &result.path);
6353 if result.removed() {
6354 count += 1;
6355 removed.push(full);
6356 } else if result.exists_after.is_none() {
6357 kept.push((
6360 full,
6361 format!(
6362 "created by this run; the tag-delete route could not confirm \
6363 it was removed (quality {})",
6364 result.quality.as_deref().unwrap_or("none")
6365 ),
6366 ));
6367 } else {
6368 kept.push((
6369 full,
6370 format!(
6371 "created by this run; the tag-delete route did not remove it \
6372 (quality {})",
6373 result.quality.as_deref().unwrap_or("none")
6374 ),
6375 ));
6376 }
6377 }
6378 events.emit_late(ProjectsEvent::TagDeleteAnswered {
6379 provider: provider.to_string(),
6380 clean: answer.is_clean(),
6381 deleted: count,
6382 });
6383 }
6384 Err(err) => {
6385 let why = format!(
6386 "created by this run; its delete call failed ({})",
6387 err.code()
6388 );
6389 kept.extend(
6390 chunk
6391 .iter()
6392 .map(|p| (full_tag_path(provider, p), why.clone())),
6393 );
6394 }
6395 }
6396 }
6397 (removed, kept)
6398}
6399
6400fn sorted_capped(mut items: Vec<String>) -> (Vec<String>, usize) {
6402 items.sort_by(|a, b| a.as_bytes().cmp(b.as_bytes()));
6403 items.dedup();
6404 capped(items)
6405}
6406
6407#[allow(clippy::too_many_arguments)]
6436async fn roll_back(
6437 failure: TagFailure,
6438 project_undo: Option<ProjectUndo>,
6439 mut result: DeployProjectsResult,
6440 loaded: &LoadedProject,
6441 input: &ProjectsInput,
6442 surfaces: &ProjectsSurfaces<'_>,
6443 opts: &ProjectsRunOptions<'_>,
6444 run_id: &RunId,
6445 rel: &str,
6446 undo_writers: UndoWriters<'_>,
6447 route: &RouteState<'_>,
6448 events: &mut RunEvents<'_>,
6449) -> Result<DeployProjectsResult, CoreError> {
6450 let run_writer = undo_writers.run;
6451 let gateway = surfaces.gateway;
6452 let TagFailure {
6453 cause,
6454 what,
6455 mut writes,
6456 } = failure;
6457 let project = result.project.clone();
6458 let backup_since_ms = wall_clock_ms();
6460 events.emit_late(ProjectsEvent::RollbackStarted { cause });
6461 journal_rolling_back(
6463 &mut result,
6464 loaded,
6465 run_writer,
6466 writes.host.as_mut(),
6467 events,
6468 )
6469 .await;
6470 let mut left: Vec<String> = Vec::new();
6471 let mut reimported: Vec<String> = Vec::new();
6472 let mut carried_back: Vec<String> = Vec::new();
6473 let mut undeletable: Vec<String> = Vec::new();
6474 let mut undeletable_left: Vec<String> = Vec::new();
6476 let mut deleted: Vec<String> = Vec::new();
6478 let mut delete_left: Vec<(String, String)> = Vec::new();
6479 let mut uncarriable_paths: Vec<String> = Vec::new();
6480 let mut recreated_tops: BTreeMap<String, Vec<String>> = BTreeMap::new();
6483 let mut project_reimported = false;
6484
6485 if let Some(undo) = &project_undo {
6487 match &undo.pre {
6488 Some(zip) => match live_digest(&project, zip) {
6489 Ok(pre) => {
6490 let now = project_sha(gateway, &project).await;
6491 if now.as_deref() != Some(pre.sha256.as_str()) {
6492 match gateway.project_import(&project, zip.clone(), true).await {
6493 Ok(_) => project_reimported = true,
6494 Err(err) => left.push(format!(
6495 "project {project}: the import of its pre-image failed ({})",
6496 err.code()
6497 )),
6498 }
6499 let back = project_sha(gateway, &project).await;
6500 let equal = back.as_deref() == Some(pre.sha256.as_str());
6501 events.emit_late(ProjectsEvent::ProjectRestored { equal });
6502 if !equal && project_reimported {
6503 left.push(format!(
6504 "project {project}: it does not read back as its pre-image \
6505 ({rel}/export/{project}.zip)"
6506 ));
6507 }
6508 }
6509 }
6510 Err(err) => left.push(format!(
6511 "project {project}: its pre-image cannot be read ({})",
6512 err.code()
6513 )),
6514 },
6515 None => match gateway.project_find(&project).await {
6516 Ok(Some(_)) => result.warnings.push(format!(
6517 "project {project} was created by this run and stays: the gateway surface \
6518 has no project delete, so the rollback cannot remove it"
6519 )),
6520 Ok(None) => {}
6521 Err(err) => result.warnings.push(format!(
6522 "project {project}: whether this run created it could not be read ({})",
6523 err.code()
6524 )),
6525 },
6526 }
6527 }
6528
6529 let empty = BTreeMap::new();
6531 for provider in sorted(&loaded.owned_tags) {
6532 let name = provider.name.as_str();
6533 if writes.created.contains_key(name) {
6534 continue;
6536 }
6537 let Some(applied) = writes.applied.get(name).filter(|a| !a.is_empty()) else {
6538 continue;
6539 };
6540 let Some(provider_plan) = writes
6541 .plan
6542 .as_ref()
6543 .and_then(|p| p.providers.iter().find(|pp| pp.provider == name))
6544 else {
6545 left.push(format!("[{name}]: no plan to roll back against"));
6546 continue;
6547 };
6548 let pre = writes.pre.get(name).unwrap_or(&empty);
6549 let current = match read_flat(gateway, name).await {
6550 Ok(current) => current,
6551 Err(err) => {
6552 left.push(format!(
6553 "[{name}]: the provider could not be read to plan its rollback ({}); every \
6554 node the run sent stays as it is",
6555 err.code()
6556 ));
6557 continue;
6558 }
6559 };
6560 let no_record = BTreeSet::new();
6561 let recreate = writes.recreate.get(name).unwrap_or(&no_record);
6562 let undo = match rollback_plan(
6563 provider_plan,
6564 pre,
6565 ¤t,
6566 applied,
6567 false,
6568 recreate,
6569 route.ready().map(|(_, handshake)| handshake),
6570 TAG_DELETE_DESCENDANTS,
6571 ) {
6572 Ok(undo) => undo,
6573 Err(err) => {
6574 left.push(format!(
6575 "[{name}]: its rollback could not be planned ({})",
6576 err.code()
6577 ));
6578 continue;
6579 }
6580 };
6581 if !undo.recreated_deleted.is_empty() {
6582 recreated_tops
6583 .entry(name.to_string())
6584 .or_default()
6585 .extend(undo.recreated_deleted.iter().cloned());
6586 }
6587 let kept = route.kept_reason(name);
6588 for path in undo
6591 .undeletable
6592 .iter()
6593 .filter(|p| !undo.created.contains(p))
6594 {
6595 let full = full_tag_path(name, path);
6596 undeletable_left.push(format!("{full} ({kept})"));
6597 undeletable.push(full);
6598 }
6599 for (path, why) in &undo.foreign {
6602 let full = full_tag_path(name, path);
6603 undeletable_left.push(format!("{full} ({})", why.describe()));
6604 undeletable.push(full);
6605 }
6606 uncarriable_paths.extend(undo.uncarriable.iter().map(|p| full_tag_path(name, p)));
6607 let full = |batch: &ImportBatch| -> Vec<String> {
6608 batch.nodes.iter().map(|p| full_tag_path(name, p)).collect()
6609 };
6610 let mut queue: VecDeque<ImportBatch> = undo.imports.into();
6611 while let Some(batch) = queue.pop_front() {
6612 let body = if batch.carry {
6613 let fresh = match read_flat(gateway, name).await {
6614 Ok(fresh) => fresh,
6615 Err(err) => {
6616 left.extend(full(&batch).into_iter().map(|p| {
6617 format!(
6618 "{p} (not restored: the export to carry its value from failed, {})",
6619 err.code()
6620 )
6621 }));
6622 continue;
6623 }
6624 };
6625 match uncarriable_in_batch(&batch, &fresh) {
6626 Ok(bad) if bad.is_empty() => {}
6627 Ok(bad) => {
6628 let bad: BTreeSet<String> = bad
6629 .into_iter()
6630 .map(|(p, _)| full_tag_path(name, &p))
6631 .collect();
6632 left.extend(full(&batch).into_iter().filter(|p| !bad.contains(p)).map(
6633 |p| format!("{p} (not restored: its batch held an uncarriable tag)"),
6634 ));
6635 uncarriable_paths.extend(bad);
6636 continue;
6637 }
6638 Err(err) => {
6639 left.extend(
6640 full(&batch)
6641 .into_iter()
6642 .map(|p| format!("{p} (not restored: {})", err.code())),
6643 );
6644 continue;
6645 }
6646 }
6647 match carried(&batch, &fresh) {
6648 Ok(Carried::Body(bytes)) => bytes,
6649 Ok(Carried::Split(parts)) => {
6650 for part in parts.into_iter().rev() {
6651 queue.push_front(part);
6652 }
6653 continue;
6654 }
6655 Err(err) => {
6656 left.extend(
6657 full(&batch)
6658 .into_iter()
6659 .map(|p| format!("{p} (not restored: {})", err.code())),
6660 );
6661 continue;
6662 }
6663 }
6664 } else {
6665 batch.body.clone()
6666 };
6667 if batch.carry {
6668 events.emit_late(ProjectsEvent::TagValuesCarried {
6669 provider: name.to_string(),
6670 path: batch.path.clone(),
6671 nodes: batch.nodes.len(),
6672 });
6673 }
6674 events.emit_late(ProjectsEvent::TagImportSent {
6675 provider: name.to_string(),
6676 path: batch.path.clone(),
6677 policy: batch.policy.as_wire(),
6678 nodes: batch.nodes.len(),
6679 carry: batch.carry,
6680 });
6681 let label = format!(
6682 "{} ({}, {} node(s))",
6683 full_tag_path(name, &batch.path),
6684 batch.policy.as_wire(),
6685 batch.nodes.len()
6686 );
6687 match gateway
6688 .tag_import(name, &batch.path, batch.policy, body)
6689 .await
6690 {
6691 Ok(answer) => {
6692 let clean = answer.is_clean();
6693 events.emit_late(ProjectsEvent::TagImportAnswered {
6694 provider: name.to_string(),
6695 clean,
6696 });
6697 if clean {
6698 reimported.extend(full(&batch));
6699 if batch.carry {
6700 carried_back.extend(full(&batch));
6701 }
6702 } else {
6703 left.push(format!("{label}: the restore import was not clean"));
6704 }
6705 }
6706 Err(err) => left.push(format!(
6707 "{label}: the restore import failed ({})",
6708 err.code()
6709 )),
6710 }
6711 }
6712 if !undo.created.is_empty() {
6717 let left_all = |why: String| -> Vec<(String, String)> {
6718 undo.created
6719 .iter()
6720 .map(|p| (full_tag_path(name, p), why.clone()))
6721 .collect()
6722 };
6723 let Some((client, handshake)) = route.ready() else {
6724 delete_left.extend(left_all(
6725 "created by this run; the tag-delete route is not ready".to_string(),
6726 ));
6727 continue;
6728 };
6729 let fresh = match read_flat(gateway, name).await {
6730 Ok(fresh) => fresh,
6731 Err(err) => {
6732 delete_left.extend(left_all(format!(
6733 "created by this run; the provider could not be read before its deletes \
6734 ({})",
6735 err.code()
6736 )));
6737 continue;
6738 }
6739 };
6740 match created_deletes_now(name, &undo.created, &fresh, applied, TAG_DELETE_DESCENDANTS)
6741 {
6742 Ok(drawn) => {
6743 let deletes = drawn.deletes;
6744 let foreign = drawn.foreign;
6745 for path in &drawn.kept {
6746 let full = full_tag_path(name, path);
6747 undeletable_left.push(format!("{full} ({kept})"));
6748 undeletable.push(full);
6749 }
6750 for (path, user) in &drawn.in_use {
6751 let full = full_tag_path(name, path);
6752 undeletable_left.push(format!(
6753 "{full} (created by this run; the live node {} still uses it, so it \
6754 was not removed)",
6755 full_tag_path(name, user)
6756 ));
6757 undeletable.push(full);
6758 }
6759 for (path, why) in &foreign {
6760 let full = full_tag_path(name, path);
6761 undeletable_left.push(format!("{full} ({})", why.describe()));
6762 undeletable.push(full);
6763 }
6764 if !deletes.is_empty() {
6765 let (removed, kept) =
6766 delete_created(client, handshake, name, &deletes, events).await;
6767 deleted.extend(removed);
6768 delete_left.extend(kept);
6769 }
6770 }
6771 Err(err) => delete_left.extend(left_all(format!(
6772 "created by this run; its deletes could not be planned ({})",
6773 err.code()
6774 ))),
6775 }
6776 }
6777 }
6778
6779 let mut groups_restored = false;
6781 match &writes.groups {
6782 Some(GroupsDone::Committed { record, providers }) => {
6783 let journal = journal_location(
6784 LedgerRoot::for_node(&loaded.target.nodes.master),
6785 &loaded.target.gateway,
6786 &loaded.target.mode,
6787 );
6788 match (surfaces.apply_host, surfaces.deploy) {
6789 (Some(host), Some(deploy)) => {
6790 match restore_groups(host, deploy, record, run_id, None).await {
6791 Ok(()) => {
6792 groups_restored = true;
6793 events.emit_late(ProjectsEvent::TagGroupsRestored {
6794 providers: providers.clone(),
6795 });
6796 }
6797 Err(why) => left.push(format!(
6798 "the tag groups of {}: {why}; see the on-box journal {journal}",
6799 providers.join(", ")
6800 )),
6801 }
6802 }
6803 _ => left.push(format!(
6804 "the tag groups of {}: no apply host to restore them with",
6805 providers.join(", ")
6806 )),
6807 }
6808 }
6809 Some(GroupsDone::Restored) => groups_restored = true,
6810 Some(GroupsDone::Unrestored(why)) => left.push(why.clone()),
6811 None => {}
6812 }
6813
6814 let mut provider_restored = Vec::new();
6816 let mut provider_deleted = Vec::new();
6817 for (name, pre) in writes.modified.iter().rev() {
6818 match restore_provider_settings(gateway, name, pre).await {
6819 Ok(true) => {
6820 provider_restored.push(name.clone());
6821 events.emit_late(ProjectsEvent::TagProviderRestored {
6822 provider: name.clone(),
6823 action: "modify",
6824 });
6825 }
6826 Ok(false) => {}
6827 Err(why) => left.push(format!(
6828 "[{name}] provider settings: {why} (pre-image {rel}/export/tags/{name}.provider.json)"
6829 )),
6830 }
6831 }
6832 for (name, created) in &writes.created {
6833 match delete_created_provider(gateway, name, created).await {
6834 Ok(true) => {
6835 provider_deleted.push(name.clone());
6836 events.emit_late(ProjectsEvent::TagProviderRestored {
6837 provider: name.clone(),
6838 action: "delete",
6839 });
6840 }
6841 Ok(false) => {}
6842 Err(why) => left.push(format!("[{name}] the provider this run created: {why}")),
6843 }
6844 }
6845
6846 let owned = sorted(&loaded.owned_tags);
6848 let names: Vec<&str> = owned
6851 .iter()
6852 .map(|p| p.name.as_str())
6853 .filter(|name| writes.pre.contains_key(*name))
6854 .collect();
6855 let exempt: BTreeSet<String> = undeletable
6856 .iter()
6857 .chain(&uncarriable_paths)
6858 .chain(delete_left.iter().map(|(path, _)| path))
6859 .cloned()
6860 .collect();
6861 let mut verify = None;
6862 if !names.is_empty() {
6863 match settled_reads(
6864 gateway,
6865 &names,
6866 opts.tag_verify_settle,
6867 opts.tag_verify_poll,
6868 )
6869 .await
6870 {
6871 Ok((last, reads, settled, elapsed_s)) => {
6872 let mut mismatches = Vec::new();
6873 for name in &names {
6874 let pre = writes.pre.get(*name).unwrap_or(&empty);
6875 let now = last.get(*name).unwrap_or(&empty);
6876 let tops = recreated_tops.get(*name).map(Vec::as_slice).unwrap_or(&[]);
6877 for (path, before) in pre {
6878 let full = full_tag_path(name, path);
6879 if exempt.contains(&full) {
6880 continue;
6881 }
6882 match now.get(path) {
6883 None => mismatches.push(format!("{full} (absent)")),
6884 Some(after) => {
6885 let opts = if recreated_at(path, tops) {
6891 RECREATED
6892 } else {
6893 EXPORTS
6894 };
6895 let properties = diff_properties(before, after, path, opts);
6896 if !properties.is_empty() {
6897 mismatches.push(format!("{full} ({})", properties.join(", ")));
6898 }
6899 }
6900 }
6901 }
6902 for path in now.keys() {
6903 let full = full_tag_path(name, path);
6904 if !pre.contains_key(path) && !exempt.contains(&full) {
6905 mismatches.push(format!("{full} (not in the pre-image)"));
6906 }
6907 }
6908 }
6909 mismatches.sort_by(|a, b| a.as_bytes().cmp(b.as_bytes()));
6910 let equal = mismatches.is_empty();
6911 events.emit_late(ProjectsEvent::RollbackVerified { equal });
6912 left.extend(mismatches.iter().map(|m| format!("verify: {m}")));
6913 let mismatches_count = mismatches.len();
6914 mismatches.truncate(LIST_CAP);
6915 verify = Some(TagVerify {
6916 equal,
6917 settled,
6918 reads,
6919 elapsed_s,
6920 mismatches,
6921 mismatches_count,
6922 });
6923 }
6924 Err(err) => left.push(format!(
6925 "the owned tag providers could not be read back to verify the rollback ({})",
6926 err.code()
6927 )),
6928 }
6929 }
6930
6931 left.extend(undeletable_left);
6932 left.extend(
6933 delete_left
6934 .iter()
6935 .map(|(path, why)| format!("{path} ({why})")),
6936 );
6937 left.extend(uncarriable_paths.iter().map(|p| {
6938 format!("{p} (its live value cannot be carried into an Overwrite, so it was not restored)")
6939 }));
6940
6941 if left.is_empty() {
6942 result.outcome = if project_undo.is_some() {
6943 ProjectOutcome::RolledBack
6944 } else {
6945 ProjectOutcome::TagsRolledBack
6946 };
6947 result.ok = false;
6948 result.slug = Some(ROLLED_BACK_SLUG.to_string());
6949 result.record_written = false;
6950 result.restore_command = None;
6951 if let Some(tags) = result.tags.as_mut() {
6952 let (carried, carried_count) = sorted_capped(writes.carried.clone());
6953 tags.carried = carried;
6954 tags.carried_count = carried_count;
6955 tags.verify = writes.verify.clone();
6956 let (reimported, reimported_count) = sorted_capped(reimported);
6957 let (carried, carried_count) = sorted_capped(carried_back);
6958 provider_restored.sort_by(|a, b| a.as_bytes().cmp(b.as_bytes()));
6959 provider_deleted.sort_by(|a, b| a.as_bytes().cmp(b.as_bytes()));
6960 let (deleted, deleted_count) = sorted_capped(deleted);
6961 tags.rollback = Some(TagRollback {
6962 cause: cause.as_str().to_string(),
6963 detail: what,
6964 reimported,
6965 reimported_count,
6966 carried,
6967 carried_count,
6968 undeletable: Vec::new(),
6969 undeletable_count: 0,
6970 groups_restored,
6971 provider_restored,
6972 provider_deleted,
6973 project_reimported,
6974 verify,
6975 deleted,
6976 deleted_count,
6977 });
6978 }
6979 let records_back = match &writes.records_back {
6982 Some(back) => {
6983 restore_tag_records(&mut result, loaded, undo_writers.records, back, events).await
6984 }
6985 None => true,
6986 };
6987 if let Some(back) = &writes.release_back {
6991 restore_release_record(&mut result, loaded, undo_writers.release, back.as_deref())
6992 .await;
6993 }
6994 if records_back {
6996 clear_host_journal(
6997 &mut result,
6998 loaded,
6999 run_writer,
7000 writes.host.as_ref(),
7001 events,
7002 )
7003 .await;
7004 }
7005 if backup_step(&mut result, &names, surfaces, opts, backup_since_ms, events).await
7009 == Some(false)
7010 {
7011 let reason = result
7012 .tags
7013 .as_ref()
7014 .and_then(|tags| tags.backup.as_ref())
7015 .and_then(|backup| backup.reason)
7016 .map_or("unknown", |reason| reason.as_str());
7017 result.warnings.push(format!(
7018 "the backup did not converge after the rollback ({reason}): the master is \
7019 restored and verified, and data.tags.backup says what the backup holds"
7020 ));
7021 }
7022 events.emit_late(ProjectsEvent::Finished {
7023 outcome: result.outcome,
7024 });
7025 note_late_log_failure(&mut result, events);
7026 return Ok(result);
7027 }
7028
7029 let listed = |items: &[String]| {
7030 let shown: Vec<&str> = items
7031 .iter()
7032 .take(TAG_FAILURE_PATHS_SHOWN)
7033 .map(String::as_str)
7034 .collect();
7035 let more = items.len().saturating_sub(shown.len());
7036 format!(
7037 "{}{}",
7038 if shown.is_empty() {
7039 "none".to_string()
7040 } else {
7041 shown.join("; ")
7042 },
7043 if more > 0 {
7044 format!(" and {more} more")
7045 } else {
7046 String::new()
7047 }
7048 )
7049 };
7050 let pre_images = match (&writes.host, cause) {
7053 (Some(_), RollbackCause::RollbackRun) => pre_image_location(
7054 LedgerRoot::for_node(&loaded.target.nodes.master),
7055 &loaded.target.gateway,
7056 &loaded.file.project,
7057 crate::deploy::project_run::PRE_IMAGE_DIR,
7058 ),
7059 _ => format!("{rel}/export"),
7060 };
7061 Err(CoreError::RollbackFailed(Box::new(RollbackFailure {
7062 run_id: run_id.as_str().to_string(),
7063 env: input.env.clone(),
7064 run_dir: Some(rel.to_string()),
7065 journal: pre_images,
7066 cause: format!(
7067 "{}: {what}; {} {} tag write(s): {}; the rollback put back what it could \
7068 and {} item(s) are left: {}{}",
7069 cause.as_str(),
7070 if cause == RollbackCause::RollbackRun {
7071 "the undone run planned"
7072 } else {
7073 "the run sent"
7074 },
7075 writes.sent.len(),
7076 listed(&writes.sent),
7077 left.len(),
7078 listed(&left),
7079 match &writes.host {
7080 Some(host) => format!(
7083 "; the host journal {} stays, and `{}` retries the undo from the host copy",
7084 host_journal_location(loaded),
7085 rollback_run_command(&input.env, &host.journal.run_id)
7086 ),
7087 None => String::new(),
7088 }
7089 ),
7090 scope: RollbackScope::Projects,
7091 })))
7092}
7093
7094pub fn rollback_run_prose(
7101 loaded: &LoadedProject,
7102 env: &str,
7103 run_id: &str,
7104 route: &TagDeleteSurface<'_>,
7105) -> String {
7106 let created = match route {
7107 TagDeleteSurface::Configured(_) => {
7108 "Tags the run created in a provider that existed are removed through the tag-delete \
7109 route when it answers and lists the provider; one the route cannot remove, or that \
7110 is not what the run wrote, stays and ends the undo as rollback_failed"
7111 }
7112 TagDeleteSurface::NotConfigured { .. } => {
7113 "Tags the run created in a provider that existed cannot be removed without the \
7114 tag-delete route, which this profile does not configure, and end the undo as \
7115 rollback_failed"
7116 }
7117 };
7118 format!(
7119 "deploy projects will undo run {run_id} of project {} on gateway {} (env {env}) from \
7120 the copy the gateway host keeps: the owned tags of {}, their tag groups, provider \
7121 settings and tag records, and the project, back to that run's pre-image. Tags the run \
7122 deleted are re-created from the pre-image. {created}. `ign deploy projects --env {env} \
7123 --check --rollback-run {run_id}` lists what it restores",
7124 loaded.file.project,
7125 loaded.target.gateway,
7126 loaded.file.tag_providers.join(", "),
7127 )
7128}
7129
7130struct HostSet {
7132 run: PreImageRun,
7133 journal: Option<ProjectJournal>,
7134 project_zip: Option<Vec<u8>>,
7135 release: Option<Vec<u8>>,
7138 pre: ProviderReads,
7139 settings: BTreeMap<String, TagProviderRecord>,
7140 records: BTreeMap<String, (TagRecord, Vec<u8>)>,
7142 sent: BTreeMap<String, BTreeMap<String, Value>>,
7145}
7146
7147async fn read_host_set(
7153 loaded: &LoadedProject,
7154 input: &ProjectsInput,
7155 host: &dyn ProjectHost,
7156 target: &RunId,
7157) -> Result<HostSet, CoreError> {
7158 let gateway = &loaded.target.gateway;
7159 let project = &loaded.file.project;
7160 let root = LedgerRoot::for_node(&loaded.target.nodes.master);
7161 let invalid = |problem: String| CoreError::DeployRunInvalid {
7162 run_id: target.as_str().to_string(),
7163 problem,
7164 };
7165 let journal = read_project_journal(loaded, host).await?;
7166 if let Some(journal) = &journal
7167 && journal.run_id != target.as_str()
7168 {
7169 return Err(project_journal_present(loaded, &input.env, &journal.run_id));
7170 }
7171 let run_location = pre_image_location(root, gateway, project, "run.json");
7172 let Some(bytes) = host.read_pre_image(gateway, project, "run.json").await? else {
7173 return Err(invalid(format!(
7174 "{run_location} does not exist: the host keeps the pre-image of the latest \
7175 tag-owning run of this project only, and holds none"
7176 )));
7177 };
7178 let run = PreImageRun::parse(&bytes, &run_location, gateway, project)?;
7179 if run.run_id != target.as_str() {
7180 return Err(invalid(format!(
7181 "the host keeps the pre-image of run {} only, the latest run of this project; \
7182 run {} cannot be undone from it",
7183 run.run_id,
7184 target.as_str()
7185 )));
7186 }
7187 for provider in &run.providers {
7188 if !loaded.file.tag_providers.contains(&provider.name) {
7189 return Err(invalid(format!(
7190 "{run_location} lists tag provider {}, which deploy/project.yaml does not own: \
7191 only the project's own providers are restored",
7192 provider.name
7193 )));
7194 }
7195 }
7196 if run.project_action == "import" && !run.files.contains_key("project.zip") {
7197 return Err(invalid(format!(
7198 "{run_location}: the run imported the project but the set holds no project.zip"
7199 )));
7200 }
7201 let mut set = HostSet {
7202 run,
7203 journal,
7204 project_zip: None,
7205 release: None,
7206 pre: BTreeMap::new(),
7207 settings: BTreeMap::new(),
7208 records: BTreeMap::new(),
7209 sent: BTreeMap::new(),
7210 };
7211 for (name, sha) in &set.run.files {
7212 let location = pre_image_location(root, gateway, project, name);
7213 let Some(bytes) = host.read_pre_image(gateway, project, name).await? else {
7214 return Err(invalid(format!(
7215 "{location} is missing: the pre-image set is incomplete"
7216 )));
7217 };
7218 if sha256_hex(&bytes) != *sha {
7219 return Err(invalid(format!(
7220 "{location} does not hash as run.json records: the set is incomplete or changed"
7221 )));
7222 }
7223 match PreImageFile::parse(name) {
7224 Some(PreImageFile::Project) => set.project_zip = Some(bytes),
7225 Some(PreImageFile::Release) => {
7226 ReleaseRecord::parse(&bytes, &location, gateway, project)?;
7227 set.release = Some(bytes);
7228 }
7229 Some(PreImageFile::Tags(provider)) => {
7230 let flat = flatten_export(&bytes, &provider)?;
7231 set.pre.insert(provider, flat);
7232 }
7233 Some(PreImageFile::Provider(provider)) => {
7234 let record: TagProviderRecord = serde_json::from_slice(&bytes)
7235 .map_err(|_| invalid(format!("{location} is not a tag provider record")))?;
7236 set.settings.insert(provider, record);
7237 }
7238 Some(PreImageFile::Record(provider)) => {
7239 let record = TagRecord::parse(&bytes, &location, gateway, &provider)?;
7240 if record.project != *project {
7241 return Err(invalid(format!(
7242 "{location} is another project's tag record"
7243 )));
7244 }
7245 set.records.insert(provider, (record, bytes));
7246 }
7247 Some(PreImageFile::Sent(provider)) => {
7248 let sent: BTreeMap<String, Value> =
7249 serde_json::from_slice(&bytes).map_err(|_| {
7250 invalid(format!(
7251 "{location} is not an object of node paths to the nodes the run sent"
7252 ))
7253 })?;
7254 set.sent.insert(provider, sent);
7255 }
7256 Some(PreImageFile::Run) | None => {
7257 return Err(invalid(format!("{location}: not a file of the set")));
7258 }
7259 }
7260 }
7261 let empty = BTreeMap::new();
7264 for provider in &set.run.providers {
7265 let pre = set.pre.get(&provider.name).unwrap_or(&empty);
7266 if let Some(path) = provider.deleted.iter().find(|p| !pre.contains_key(*p)) {
7267 return Err(invalid(format!(
7268 "{run_location} names {} among the paths run {} deleted, and the set's \
7269 pre-image of the provider does not hold it: the set is stale or partial",
7270 full_tag_path(&provider.name, path),
7271 target.as_str()
7272 )));
7273 }
7274 }
7275 Ok(set)
7276}
7277
7278fn undo_writes(set: &HostSet) -> TagWrites {
7284 let mut writes = TagWrites::default();
7285 let mut providers = Vec::new();
7286 let mut records_back = Vec::new();
7287 for provider in &set.run.providers {
7288 let name = provider.name.clone();
7289 providers.push(ProviderPlan {
7290 provider: name.clone(),
7291 added: provider.added.clone(),
7292 recreated: Vec::new(),
7293 changed: provider
7294 .changed
7295 .iter()
7296 .map(|path| NodeChange {
7297 path: path.clone(),
7298 properties: Vec::new(),
7299 restart_likely: false,
7300 })
7301 .collect(),
7302 kept_values: Vec::new(),
7303 removed: provider.removed.clone(),
7304 already_absent: Vec::new(),
7305 drift: Vec::new(),
7306 same: 0,
7307 first_change: false,
7308 record_only: false,
7309 provider_emptied: false,
7310 record_run_id: None,
7311 });
7312 let mut applied = AppliedSteps::default();
7313 match &provider.sent {
7317 Some(sent) => applied.nodes.extend(sent.iter().cloned()),
7318 None => {
7319 applied.nodes.extend(provider.added.iter().cloned());
7320 applied.nodes.extend(provider.changed.iter().cloned());
7321 }
7322 }
7323 if let Some(bodies) = set.sent.get(&name) {
7324 applied.bodies.extend(
7325 bodies
7326 .iter()
7327 .filter(|(path, _)| applied.nodes.contains(*path))
7328 .map(|(path, node)| (path.clone(), node.clone())),
7329 );
7330 }
7331 applied.collided.extend(provider.collided.iter().cloned());
7332 applied.deleted.clone_from(&provider.deleted);
7335 if !applied.is_empty() {
7336 writes.sent.push(format!(
7337 "[{name}] {} planned node(s), {} deleted path(s)",
7338 applied.nodes.len(),
7339 applied.deleted.len()
7340 ));
7341 writes.applied.insert(name.clone(), applied);
7342 }
7343 if let Some(proof) = &provider.created {
7344 writes
7345 .sent
7346 .push(format!("[{name}] provider settings (create)"));
7347 writes.created.insert(
7348 name.clone(),
7349 CreatedProvider {
7350 confirmed: proof.confirmed,
7351 sent: (
7352 proof.enabled,
7353 proof.description.clone(),
7354 proof.config.clone(),
7355 ),
7356 signature: proof.signature.clone(),
7357 },
7358 );
7359 }
7360 if provider.settings == "modify"
7361 && let Some(record) = set.settings.get(&name)
7362 {
7363 writes
7364 .sent
7365 .push(format!("[{name}] provider settings (modify)"));
7366 writes.modified.push((name.clone(), record.clone()));
7367 }
7368 match set.records.get(&name) {
7369 Some((record, bytes)) => {
7370 writes
7371 .recreate
7372 .insert(name.clone(), record.nodes.iter().cloned().collect());
7373 records_back.push((name, Some(bytes.clone())));
7374 }
7375 None => records_back.push((name, None)),
7376 }
7377 }
7378 writes.plan = Some(TagPlan {
7379 providers,
7380 tags_sha256: String::new(),
7381 live_sha256: String::new(),
7382 plan_sha256: String::new(),
7383 });
7384 writes.pre = set.pre.clone();
7385 writes.records_back = Some(records_back);
7386 writes.release_back = Some(set.release.clone());
7387 writes
7388}
7389
7390async fn undo_groups(
7395 set: &HostSet,
7396 input: &ProjectsInput,
7397 apply_host: Option<&dyn ApplyHost>,
7398 target: &RunId,
7399) -> Option<GroupsDone> {
7400 if !set.run.groups {
7401 return None;
7402 }
7403 let providers: Vec<String> = set.run.providers.iter().map(|p| p.name.clone()).collect();
7404 let recover = format!(
7405 "the tag groups of {}: the group commit of run {} did not finish on the box; recover \
7406 it with `ign deploy rollback --run {} --env {}` first",
7407 providers.join(", "),
7408 target.as_str(),
7409 target.as_str(),
7410 shell_word(&input.env)
7411 );
7412 let Some(host) = apply_host else {
7413 return Some(GroupsDone::Unrestored(format!(
7414 "the tag groups of {}: no apply host to restore them with",
7415 providers.join(", ")
7416 )));
7417 };
7418 match host.read_record(target).await {
7419 Ok(None) => None,
7420 Ok(Some(record)) => match record.state {
7421 RecordState::RolledBack => None,
7422 RecordState::Applied => match live_changes(host, &record).await {
7426 Ok(changed) if changed.is_empty() => Some(GroupsDone::Committed {
7427 record: Box::new(record),
7428 providers,
7429 }),
7430 Ok(changed) => Some(GroupsDone::Unrestored(format!(
7431 "the tag groups of {}: {} unit file(s) changed since run {} ({}{}); they were \
7432 not put back over that change",
7433 providers.join(", "),
7434 changed.len(),
7435 target.as_str(),
7436 changed
7437 .iter()
7438 .take(APPLY_LIST_SHOWN)
7439 .cloned()
7440 .collect::<Vec<_>>()
7441 .join(", "),
7442 if changed.len() > APPLY_LIST_SHOWN {
7443 ", ..."
7444 } else {
7445 ""
7446 }
7447 ))),
7448 Err(err) => Some(GroupsDone::Unrestored(format!(
7449 "the tag groups of {}: their unit files could not be compared with run {} ({})",
7450 providers.join(", "),
7451 target.as_str(),
7452 err.code()
7453 ))),
7454 },
7455 RecordState::Applying => Some(GroupsDone::Unrestored(recover)),
7456 },
7457 Err(err) => Some(GroupsDone::Unrestored(format!(
7458 "the tag groups of {}: the run record of run {} could not be read ({})",
7459 providers.join(", "),
7460 target.as_str(),
7461 err.code()
7462 ))),
7463 }
7464}
7465
7466async fn restore_release_record(
7471 result: &mut DeployProjectsResult,
7472 loaded: &LoadedProject,
7473 writer: Option<&dyn ReleaseRecordWriter>,
7474 back: Option<&[u8]>,
7475) {
7476 let (Some(writer), Some(bytes)) = (writer, back) else {
7477 result.warnings.push(format!(
7478 "the release record of {} was not put back: the host copy holds none from before \
7479 the run, so the record may still name the undone release",
7480 loaded.file.project
7481 ));
7482 return;
7483 };
7484 if let Err(err) = writer
7485 .write_release_record(&loaded.target.gateway, &loaded.file.project, bytes)
7486 .await
7487 {
7488 result.warnings.push(format!(
7489 "the release record of {} could not be put back ({}); it may still name the \
7490 undone release",
7491 loaded.file.project,
7492 err.code()
7493 ));
7494 }
7495}
7496
7497async fn restore_tag_records(
7502 result: &mut DeployProjectsResult,
7503 loaded: &LoadedProject,
7504 writer: Option<&dyn TagRecordWriter>,
7505 back: &[(String, Option<Vec<u8>>)],
7506 events: &mut RunEvents<'_>,
7507) -> bool {
7508 let mut ok = true;
7509 for (provider, bytes) in back {
7510 let Some(bytes) = bytes else {
7511 result.warnings.push(format!(
7512 "tag provider {provider} had no tag record before the undone run: the record \
7513 that run left, if any, stays (this release cannot remove one)"
7514 ));
7515 continue;
7516 };
7517 let Some(writer) = writer else {
7518 ok = false;
7519 continue;
7520 };
7521 match writer
7522 .write_tag_record(&loaded.target.gateway, provider, bytes)
7523 .await
7524 {
7525 Ok(()) => events.emit_late(ProjectsEvent::TagRecordWritten {
7526 provider: provider.clone(),
7527 }),
7528 Err(err) => {
7529 ok = false;
7530 events.emit_late(ProjectsEvent::TagRecordFailed {
7531 provider: provider.clone(),
7532 code: err.code().to_string(),
7533 });
7534 result.warnings.push(format!(
7535 "the tag record of {provider} could not be put back ({}); the host journal \
7536 {} stays",
7537 err.code(),
7538 host_journal_location(loaded)
7539 ));
7540 }
7541 }
7542 }
7543 ok
7544}
7545
7546fn action_of(word: &str) -> ProjectAction {
7548 match word {
7549 "create" => ProjectAction::Create,
7550 "import" => ProjectAction::Import,
7551 "record" => ProjectAction::Record,
7552 _ => ProjectAction::Skip,
7553 }
7554}
7555
7556fn rollback_run_result(
7559 loaded: &LoadedProject,
7560 input: &ProjectsInput,
7561 set: &HostSet,
7562 writes: &TagWrites,
7563 live_sha256: Option<String>,
7564 check: bool,
7565) -> DeployProjectsResult {
7566 let empty = TagPlan {
7567 providers: Vec::new(),
7568 tags_sha256: String::new(),
7569 live_sha256: String::new(),
7570 plan_sha256: String::new(),
7571 };
7572 let mut report = TagsReport::from_plan(
7573 writes.plan.as_ref().unwrap_or(&empty),
7574 &BTreeMap::new(),
7575 LIST_CAP,
7576 );
7577 report.journal = set.journal.as_ref().map(|j| TagJournal {
7578 run_id: j.run_id.clone(),
7579 state: j.state.as_str().to_string(),
7580 started_at: j.started_at.clone(),
7581 });
7582 let restored_sha = set
7583 .project_zip
7584 .as_ref()
7585 .and_then(|zip| live_digest(&loaded.file.project, zip).ok())
7586 .map(|d| d.sha256);
7587 DeployProjectsResult {
7588 env: input.env.clone(),
7589 gateway: loaded.target.gateway.clone(),
7590 project: loaded.file.project.clone(),
7591 release: set.run.release.clone(),
7592 check,
7593 action: action_of(&set.run.project_action),
7594 would_change: true,
7595 artifact_sha256: restored_sha
7596 .or_else(|| live_sha256.clone())
7597 .unwrap_or_default(),
7598 live_sha256,
7599 members: MemberDelta::default(),
7600 removed: Vec::new(),
7601 removed_count: 0,
7602 excluded: Vec::new(),
7603 excluded_count: 0,
7604 unmanaged: Vec::new(),
7605 unmanaged_count: 0,
7606 parent: loaded.meta.parent.clone(),
7607 previous_release: None,
7608 recorded_sha256: None,
7609 live_matches_last_deploy: None,
7610 allow_downgrade: false,
7611 allow_first_change: None,
7612 warnings: Vec::new(),
7613 run_id: None,
7614 run_dir: None,
7615 record_written: false,
7616 restore_command: None,
7617 children: Vec::new(),
7618 missing_tags: None,
7619 missing_tags_count: None,
7620 missing_tags_unknown_providers: None,
7621 missing_tags_unknown_providers_count: None,
7622 missing_tags_unavailable: None,
7623 missing_tags_limit: MISSING_TAGS_LIMIT,
7624 tags: Some(Box::new(report)),
7625 outcome: if check {
7626 ProjectOutcome::Checked
7627 } else {
7628 ProjectOutcome::RolledBack
7629 },
7630 ok: true,
7631 slug: None,
7632 }
7633}
7634
7635#[allow(clippy::too_many_arguments)]
7641async fn rollback_run_preview(
7642 gateway: &dyn ProjectGateway,
7643 loaded: &LoadedProject,
7644 set: &HostSet,
7645 writes: &TagWrites,
7646 groups: Option<&GroupsDone>,
7647 target: &RunId,
7648 route: &RouteState<'_>,
7649 result: &mut DeployProjectsResult,
7650) {
7651 let project = loaded.file.project.as_str();
7652 let mut reimported = Vec::new();
7653 let mut carried = Vec::new();
7654 let mut undeletable = Vec::new();
7655 let mut deleted = Vec::new();
7656 let mut project_reimported = false;
7657 if let Some(zip) = &set.project_zip
7658 && let Ok(pre) = live_digest(project, zip)
7659 {
7660 project_reimported = result.live_sha256.as_deref() != Some(pre.sha256.as_str());
7661 }
7662 if set.run.project_action == "create" && result.live_sha256.is_some() {
7663 result.warnings.push(format!(
7664 "project {project} was created by run {} and stays: the gateway surface has no \
7665 project delete",
7666 target.as_str()
7667 ));
7668 }
7669 let empty = BTreeMap::new();
7670 let no_record = BTreeSet::new();
7671 for provider in &set.run.providers {
7672 let name = provider.name.as_str();
7673 if writes.created.contains_key(name) {
7674 continue;
7675 }
7676 let (Some(applied), Some(plan)) = (
7677 writes.applied.get(name),
7678 writes
7679 .plan
7680 .as_ref()
7681 .and_then(|p| p.providers.iter().find(|pp| pp.provider == name)),
7682 ) else {
7683 continue;
7684 };
7685 let current = match read_flat(gateway, name).await {
7686 Ok(current) => current,
7687 Err(err) => {
7688 result.warnings.push(format!(
7689 "[{name}]: the provider could not be read to plan its undo ({})",
7690 err.code()
7691 ));
7692 continue;
7693 }
7694 };
7695 let pre = set.pre.get(name).unwrap_or(&empty);
7696 let recreate = writes.recreate.get(name).unwrap_or(&no_record);
7697 match rollback_plan(
7698 plan,
7699 pre,
7700 ¤t,
7701 applied,
7702 false,
7703 recreate,
7704 route.ready().map(|(_, handshake)| handshake),
7705 TAG_DELETE_DESCENDANTS,
7706 ) {
7707 Ok(undo) => {
7708 for batch in &undo.imports {
7709 reimported.extend(batch.nodes.iter().map(|p| full_tag_path(name, p)));
7710 }
7711 carried.extend(undo.carried.iter().map(|p| full_tag_path(name, p)));
7712 undeletable.extend(undo.undeletable.iter().map(|p| full_tag_path(name, p)));
7713 for (path, why) in &undo.foreign {
7714 undeletable.push(full_tag_path(name, path));
7715 result.warnings.push(format!(
7716 "{}: {}",
7717 full_tag_path(name, path),
7718 why.describe()
7719 ));
7720 }
7721 deleted.extend(undo.deletes.iter().map(|p| full_tag_path(name, p)));
7722 for path in &undo.uncarriable {
7723 result.warnings.push(format!(
7724 "{}: its live value cannot be carried into an Overwrite, so the undo \
7725 would leave it and end rollback_failed",
7726 full_tag_path(name, path)
7727 ));
7728 }
7729 }
7730 Err(err) => result.warnings.push(format!(
7731 "[{name}]: its undo could not be planned ({})",
7732 err.code()
7733 )),
7734 }
7735 }
7736 let mut provider_restored = Vec::new();
7737 for (name, pre) in &writes.modified {
7738 if let Ok(Some(now)) = gateway.tag_provider_record(name).await
7739 && !settings_equal(&now, pre)
7740 {
7741 provider_restored.push(name.clone());
7742 }
7743 }
7744 let mut provider_deleted = Vec::new();
7745 for name in writes.created.keys() {
7746 if let Ok(Some(_)) = gateway.tag_provider_record(name).await {
7747 provider_deleted.push(name.clone());
7748 }
7749 }
7750 if let Some(GroupsDone::Unrestored(why)) = groups {
7751 result.warnings.push(why.clone());
7752 }
7753 if !undeletable.is_empty() {
7754 result.warnings.push(format!(
7755 "the undo would end rollback_failed: {} node(s) run {} created cannot be removed \
7756 (data.tags.delete_route says whether the tag-delete route can be used)",
7757 undeletable.len(),
7758 target.as_str()
7759 ));
7760 }
7761 if let Some(back) = &writes.records_back {
7762 for (provider, bytes) in back {
7763 if bytes.is_none() {
7764 result.warnings.push(format!(
7765 "tag provider {provider} had no tag record before run {}: the record that \
7766 run left, if any, would stay",
7767 target.as_str()
7768 ));
7769 }
7770 }
7771 }
7772 let (reimported, reimported_count) = sorted_capped(reimported);
7773 let (carried, carried_count) = sorted_capped(carried);
7774 let (undeletable, undeletable_count) = sorted_capped(undeletable);
7775 let (deleted, deleted_count) = sorted_capped(deleted);
7776 provider_restored.sort_by(|a, b| a.as_bytes().cmp(b.as_bytes()));
7777 result.would_change = project_reimported
7778 || reimported_count > 0
7779 || deleted_count > 0
7780 || !provider_restored.is_empty()
7781 || !provider_deleted.is_empty()
7782 || matches!(groups, Some(GroupsDone::Committed { .. }));
7783 if let Some(tags) = result.tags.as_mut() {
7784 tags.rollback = Some(TagRollback {
7785 cause: RollbackCause::RollbackRun.as_str().to_string(),
7786 detail: format!(
7787 "what the undo of run {} from the host copy would restore; nothing was written",
7788 target.as_str()
7789 ),
7790 reimported,
7791 reimported_count,
7792 carried,
7793 carried_count,
7794 undeletable,
7795 undeletable_count,
7796 groups_restored: matches!(groups, Some(GroupsDone::Committed { .. })),
7797 provider_restored,
7798 provider_deleted,
7799 project_reimported,
7800 verify: None,
7801 deleted,
7802 deleted_count,
7803 });
7804 }
7805}
7806
7807pub async fn deploy_projects_rollback_run(
7839 loaded: &LoadedProject,
7840 input: &ProjectsInput,
7841 surfaces: &ProjectsSurfaces<'_>,
7842 opts: &ProjectsRunOptions<'_>,
7843 run_id: &str,
7844 check: bool,
7845 observer: &dyn ProjectsObserver,
7846) -> Result<DeployProjectsResult, CoreError> {
7847 let target = RunId::parse(run_id)?;
7848 if loaded.owned_tags.is_empty() {
7849 return Err(CoreError::InvalidInput {
7850 reason: "--rollback-run undoes a tag-owning run, and deploy/project.yaml lists no \
7851 tag_providers"
7852 .to_string(),
7853 });
7854 }
7855 let project = loaded.file.project.as_str();
7856 let gateway = surfaces.gateway;
7857 let git = surfaces.host.git_entry(project).await?;
7860 if git.is_present() {
7861 return Err(CoreError::ProjectGitManaged {
7862 project: project.to_string(),
7863 location: git_entry_location(project),
7864 kind: git.as_str().to_string(),
7865 });
7866 }
7867 let set = read_host_set(loaded, input, surfaces.host, &target).await?;
7868 let mut writes = undo_writes(&set);
7869 let project_undo = match set.run.project_action.as_str() {
7870 "import" => Some(ProjectUndo {
7871 pre: set.project_zip.clone(),
7872 }),
7873 "create" => Some(ProjectUndo { pre: None }),
7874 _ => None,
7875 };
7876 let live_sha256 = project_sha(gateway, project).await;
7877
7878 let built_host;
7881 let apply_host: Option<&dyn ApplyHost> = match surfaces.apply_host {
7882 Some(host) => Some(host),
7883 None if set.run.groups => {
7884 built_host = apply_host_for_node(
7885 &opts.targets_root,
7886 &opts.target_file,
7887 "nodes.master",
7888 &opts.master,
7889 &loaded.target.gateway,
7890 &loaded.target.mode,
7891 )?;
7892 Some(built_host.as_ref())
7893 }
7894 None => None,
7895 };
7896 writes.groups = undo_groups(&set, input, apply_host, &target).await;
7897 let mut result = rollback_run_result(loaded, input, &set, &writes, live_sha256, check);
7898 let route = match &surfaces.tag_delete {
7903 TagDeleteSurface::Configured(client) => {
7904 let answer = if check {
7905 client.handshake().await
7906 } else {
7907 wait_ready(*client, opts.tag_delete_ready, opts.tag_delete_poll).await
7908 };
7909 RouteState::of(&surfaces.tag_delete, Some(answer))
7910 }
7911 surface => RouteState::of(surface, None),
7912 };
7913 if let Some(tags) = result.tags.as_mut() {
7914 tags.delete_route = Some(route.report(&owned_names(loaded)));
7915 }
7916 if let (Some(_), None) = (route.route, &route.handshake) {
7917 result.warnings.push(format!(
7918 "the tag-delete route is not ready ({}); the undo cannot remove the tags run {} \
7919 created",
7920 route.problem.as_deref().unwrap_or("no handshake"),
7921 target.as_str()
7922 ));
7923 }
7924
7925 if check {
7926 rollback_run_preview(
7927 gateway,
7928 loaded,
7929 &set,
7930 &writes,
7931 writes.groups.as_ref(),
7932 &target,
7933 &route,
7934 &mut result,
7935 )
7936 .await;
7937 return Ok(result);
7938 }
7939
7940 let built_run;
7942 let run_writer: &dyn ProjectRunWriter = match opts.run_writer {
7943 Some(writer) => writer,
7944 None => {
7945 built_run = project_run_writer_for_node(
7946 &opts.targets_root,
7947 &opts.target_file,
7948 "nodes.master",
7949 &opts.master,
7950 )?;
7951 built_run.as_ref()
7952 }
7953 };
7954 let built_tags;
7955 let tag_writer: &dyn TagRecordWriter = match opts.tag_record_writer {
7956 Some(writer) => writer,
7957 None => {
7958 built_tags = tag_record_writer_for_node(
7959 &opts.targets_root,
7960 &opts.target_file,
7961 "nodes.master",
7962 &opts.master,
7963 )?;
7964 built_tags.as_ref()
7965 }
7966 };
7967 let built_release;
7968 let release_writer: &dyn ReleaseRecordWriter = match opts.record_writer {
7969 Some(writer) => writer,
7970 None => {
7971 built_release = release_writer_for_node(
7972 &opts.targets_root,
7973 &opts.target_file,
7974 "nodes.master",
7975 &opts.master,
7976 )?;
7977 built_release.as_ref()
7978 }
7979 };
7980 wrong_target_check(loaded, gateway, opts).await?;
7982 let designers = if project_undo.as_ref().is_some_and(|undo| undo.pre.is_some()) {
7984 designers_check(&result, gateway, &opts.designers, observer).await?
7985 } else {
7986 DesignersFound {
7987 mode: "skipped",
7988 timeout_s: None,
7989 polls: Vec::new(),
7990 seen: Vec::new(),
7991 proceeded: false,
7992 skipped: true,
7993 }
7994 };
7995 let (rid, run_dir) = RunDir::create_fresh(&opts.repo, &loaded.target.gateway)?;
7996 let rel = run_dir_rel(&loaded.target.gateway, &rid);
7997 result.run_id = Some(rid.as_str().to_string());
7998 result.run_dir = Some(rel.clone());
7999 let surfaces = &ProjectsSurfaces {
8000 gateway: surfaces.gateway,
8001 host: surfaces.host,
8002 tree: surfaces.tree,
8003 apply_host,
8004 deploy: surfaces.deploy,
8005 tag_delete: surfaces.tag_delete.clone(),
8006 };
8007 let outcome = async {
8008 run_dir.write_json(
8009 "run.json",
8010 &serde_json::json!({
8011 "kind": "rollback_run",
8012 "run_id": rid.as_str(),
8013 "target_run_id": target.as_str(),
8014 "started_at": rfc3339_now_utc(),
8015 "env": input.env,
8016 "gateway": loaded.target.gateway,
8017 "project": project,
8018 "release": set.run.release,
8019 "tag_providers": loaded.file.tag_providers,
8020 "designers": designers.mode,
8021 "designers_seen": sessions_json(&designers.seen),
8022 "run_dir_git": run_dir.git_status().as_record(),
8023 }),
8024 )?;
8025 let mut events = RunEvents {
8026 log: EventLog::open(&run_dir)?,
8027 observer,
8028 late_failure: false,
8029 };
8030 events.emit(ProjectsEvent::RunStarted {
8031 run_id: rid.as_str().to_string(),
8032 action: result.action,
8033 })?;
8034 if designers.proceeded {
8035 events.emit(ProjectsEvent::DesignersProceed {
8036 sessions: designers.seen.clone(),
8037 })?;
8038 }
8039 let journal = ProjectJournal {
8042 schema_version: 1,
8043 gateway: loaded.target.gateway.clone(),
8044 project: project.to_string(),
8045 run_id: target.as_str().to_string(),
8046 state: JournalState::RollingBack,
8047 started_at: rfc3339_now_utc(),
8048 providers: set.run.providers.iter().map(|p| p.name.clone()).collect(),
8049 };
8050 run_writer
8051 .write_journal(&loaded.target.gateway, project, &journal.to_bytes())
8052 .await?;
8053 events.emit(ProjectsEvent::JournalWritten {
8054 state: JournalState::RollingBack.as_str(),
8055 })?;
8056 writes.host = Some(HostCopy {
8057 run: set.run.clone(),
8058 journal,
8059 });
8060 let failure = TagFailure {
8061 cause: RollbackCause::RollbackRun,
8062 what: format!("the undo of run {} from the host copy", target.as_str()),
8063 writes,
8064 };
8065 roll_back(
8066 failure,
8067 project_undo,
8068 result,
8069 loaded,
8070 input,
8071 surfaces,
8072 opts,
8073 &rid,
8074 &rel,
8075 UndoWriters {
8076 run: Some(run_writer),
8077 records: Some(tag_writer),
8078 release: Some(release_writer),
8079 },
8080 &route,
8081 &mut events,
8082 )
8083 .await
8084 }
8085 .await;
8086 match outcome {
8087 Ok(mut result) => {
8088 if let Err(err) = run_dir.write_json("result.json", &result) {
8089 result.warnings.push(format!(
8090 "the run's result.json could not be written ({})",
8091 err.code()
8092 ));
8093 }
8094 Ok(result)
8095 }
8096 Err(err) => {
8097 let _ = run_dir.write_json(
8098 "result.json",
8099 &serde_json::json!({
8100 "ok": false,
8101 "slug": err.code(),
8102 "run_id": rid.as_str(),
8103 "target_run_id": target.as_str(),
8104 }),
8105 );
8106 Err(err)
8107 }
8108 }
8109}
8110
8111const GROUP_SCAN_POLL: Duration = Duration::from_millis(500);
8113
8114const GROUP_SCAN_WAIT: Duration = Duration::from_secs(120);
8118
8119fn group_scan_poll() -> crate::poll::PollConfig {
8120 crate::poll::PollConfig {
8121 subject: "the config scan after the tag-group commit".to_string(),
8122 interval: GROUP_SCAN_POLL,
8123 max: Duration::from_secs(2),
8124 deadline: GROUP_SCAN_WAIT,
8125 }
8126}
8127
8128struct GroupWork {
8130 providers: Vec<String>,
8132 plans: Vec<GroupPlan>,
8134}
8135
8136impl GroupWork {
8137 fn would_change(&self) -> bool {
8138 self.plans.iter().any(GroupPlan::would_change)
8139 }
8140}
8141
8142async fn fresh_groups(
8147 loaded: &LoadedProject,
8148 input: &ProjectsInput,
8149 surfaces: &ProjectsSurfaces<'_>,
8150 report: &TagsReport,
8151 tag_plan: &TagPlan,
8152 tag_records: &std::collections::BTreeMap<String, TagRecord>,
8153 live: &ProviderReads,
8154) -> Result<Option<GroupWork>, CoreError> {
8155 if report.groups.is_empty() {
8156 return Ok(None);
8157 }
8158 let reader = need_tree(surfaces, "the tag-group write")?;
8159 let inputs = group_inputs(loaded, tag_plan)?;
8160 let providers: Vec<String> = report.groups.iter().map(|g| g.provider.clone()).collect();
8161 let observed = read_groups(reader, &loaded.target.mode, &providers).await?;
8162 let plans = group_plans(&inputs, &observed, tag_records, live)?;
8163 if plans != report.groups {
8164 let shown = |plans: &[GroupPlan]| {
8165 plans
8166 .iter()
8167 .map(|p| {
8168 format!(
8169 "{}: {} added, {} changed, {} removed",
8170 p.provider,
8171 p.added.len(),
8172 p.changed.len(),
8173 p.removed.len()
8174 )
8175 })
8176 .collect::<Vec<_>>()
8177 .join("; ")
8178 };
8179 return Err(CoreError::PlanStale {
8180 expected: shown(&report.groups),
8181 actual: shown(&plans),
8182 detail: "the owned providers' tag groups changed since the plan".to_string(),
8183 });
8184 }
8185 let work = GroupWork { providers, plans };
8186 if work.would_change() {
8187 let gateway = &loaded.target.gateway;
8188 let mode = &loaded.target.mode;
8189 if let Some(bytes) = reader.read_journal(gateway, mode).await? {
8190 let location = journal_location(
8191 LedgerRoot::for_node(&loaded.target.nodes.master),
8192 gateway,
8193 mode,
8194 );
8195 let journal = Journal::parse(&location, &bytes, gateway, mode)?;
8196 return Err(CoreError::DeployJournalPresent {
8197 run_id: journal.run_id,
8198 env: input.env.clone(),
8199 location,
8200 recover_with: None,
8201 });
8202 }
8203 }
8204 Ok(Some(work))
8205}
8206
8207fn group_batch_spec(
8211 loaded: &LoadedProject,
8212 run_id: &RunId,
8213 plan_sha256: &str,
8214 plans: &[GroupPlan],
8215) -> (BatchSpec, RenderedTree) {
8216 let mut tree = RenderedTree::default();
8217 let mut writes: Vec<WriteOp> = Vec::new();
8218 let mut deletes: Vec<DeleteOp> = Vec::new();
8219 for plan in plans {
8220 for write in &plan.writes {
8221 writes.push(WriteOp {
8222 path: write.path.clone(),
8223 sha256: sha256_hex(&write.bytes),
8224 pre: write.pre.clone(),
8225 });
8226 tree.files.insert(write.path.clone(), write.bytes.clone());
8227 }
8228 for delete in &plan.deletes {
8229 deletes.push(DeleteOp {
8230 path: delete.path.clone(),
8231 pre: delete.pre.clone(),
8232 });
8233 }
8234 }
8235 writes.sort_by(|a, b| a.path.as_str().as_bytes().cmp(b.path.as_str().as_bytes()));
8236 deletes.sort_by(|a, b| a.path.as_str().as_bytes().cmp(b.path.as_str().as_bytes()));
8237 let mut touched: Vec<TouchedEntry> = writes
8238 .iter()
8239 .map(|w| TouchedEntry {
8240 path: w.path.as_str().to_string(),
8241 pre_sha256: w.pre.clone(),
8242 post_sha256: Some(w.sha256.clone()),
8243 })
8244 .chain(deletes.iter().map(|d| TouchedEntry {
8245 path: d.path.as_str().to_string(),
8246 pre_sha256: Some(d.pre.clone()),
8247 post_sha256: None,
8248 }))
8249 .collect();
8250 touched.sort_by(|a, b| a.path.as_bytes().cmp(b.path.as_bytes()));
8251 let pre_fingerprint = fingerprint(
8252 &touched
8253 .iter()
8254 .map(|e| (e.path.clone(), e.pre_sha256.clone()))
8255 .collect(),
8256 );
8257 let gateway = loaded.target.gateway.clone();
8258 let mode = loaded.target.mode.clone();
8259 let run = run_id.as_str().to_string();
8260 let budget = HOLD_S.saturating_sub(LOCK_MARGIN_S.unsigned_abs()).max(1);
8261 let spec = BatchSpec {
8262 run_id: run_id.clone(),
8263 gateway: gateway.clone(),
8264 mode: mode.clone(),
8265 writes,
8266 deletes,
8267 ledger_new: None,
8268 ledger_pre: None,
8269 journal: Journal {
8270 schema_version: 1,
8271 run_id: run.clone(),
8272 gateway: gateway.clone(),
8273 mode: mode.clone(),
8274 phase: JournalPhase::Snapshotting,
8275 pre_fingerprint: pre_fingerprint.clone(),
8276 touched: touched.iter().map(|e| e.path.clone()).collect(),
8277 },
8278 record: RunRecord {
8279 schema_version: 1,
8280 run_id: run,
8281 gateway,
8282 mode,
8283 render_sha256: tree.render_sha256(),
8284 plan_sha256: plan_sha256.to_string(),
8285 state: RecordState::Applying,
8286 rolled_back_by: None,
8287 touched,
8288 pre_fingerprint,
8289 post_fingerprint: None,
8290 },
8291 commit_budget_s: Some(u32::try_from(budget).unwrap_or(u32::MAX)),
8292 };
8293 (spec, tree)
8294}
8295
8296enum GroupFailure {
8298 NotWritten(CoreError),
8300 Stale,
8303 Restored(String),
8305 Unrestored(String),
8307}
8308
8309async fn restore_groups(
8314 host: &dyn ApplyHost,
8315 deploy: &dyn DeployApi,
8316 record: &RunRecord,
8317 run_id: &RunId,
8318 lock: Option<&HeldLock>,
8319) -> Result<(), String> {
8320 let held = match lock {
8321 Some(lock) => matches!(
8322 lock.check(deploy, LOCK_MARGIN_S, &SystemClock).await,
8323 Ok(LockVerdict::Held { .. })
8324 ),
8325 None => false,
8326 };
8327 if !held {
8328 HeldLock::acquire(deploy, ScanKind::Config, HOLD_S, &SystemClock)
8329 .await
8330 .map_err(|e| format!("taking the scan-lock for the restore: {}", e.code()))?;
8331 }
8332 let report = host
8333 .restore(record)
8334 .await
8335 .map_err(|e| format!("restore: {}", e.code()))?;
8336 if let Some(failure) = report.failure {
8337 return Err(format!("restore: {failure}"));
8338 }
8339 match scan_and_wait(deploy, ScanKind::Config, group_scan_poll()).await {
8340 Ok(ScanOutcome::Complete(_)) => {}
8341 Ok(ScanOutcome::NotComplete { .. }) => {
8342 return Err("the scan after the restore did not complete".to_string());
8343 }
8344 Err(e) => return Err(format!("the scan after the restore: {}", e.code())),
8345 }
8346 let mut rolled_back = record.clone();
8347 rolled_back.state = RecordState::RolledBack;
8348 rolled_back.rolled_back_by = Some(run_id.as_str().to_string());
8349 rolled_back.post_fingerprint = Some(record.pre_fingerprint.clone());
8350 host.finalize(&rolled_back)
8351 .await
8352 .map_err(|e| format!("finalize: {}", e.code()))
8353}
8354
8355async fn restored(
8357 host: &dyn ApplyHost,
8358 deploy: &dyn DeployApi,
8359 record: &RunRecord,
8360 run_id: &RunId,
8361 lock: &HeldLock,
8362 why: String,
8363) -> GroupFailure {
8364 match restore_groups(host, deploy, record, run_id, Some(lock)).await {
8365 Ok(()) => GroupFailure::Restored(why),
8366 Err(cause) => GroupFailure::Unrestored(format!("{why}; {cause}")),
8367 }
8368}
8369
8370async fn commit_groups(
8377 loaded: &LoadedProject,
8378 surfaces: &ProjectsSurfaces<'_>,
8379 run_id: &RunId,
8380 plan_sha256: &str,
8381 work: &GroupWork,
8382) -> Result<(usize, usize, RunRecord), GroupFailure> {
8383 let missing = |what: &str| {
8384 GroupFailure::NotWritten(CoreError::Internal(format!(
8385 "the tag-group write needs the master's {what}, and none was supplied"
8386 )))
8387 };
8388 let host = surfaces.apply_host.ok_or_else(|| missing("apply host"))?;
8389 let deploy = surfaces.deploy.ok_or_else(|| missing("deploy API"))?;
8390 let reader = surfaces.tree.ok_or_else(|| missing("host reader"))?;
8391 let (spec, tree) = group_batch_spec(loaded, run_id, plan_sha256, &work.plans);
8392 let touched = spec.touched_paths();
8393 let probe = host
8394 .probe_paths(&touched)
8395 .await
8396 .map_err(GroupFailure::NotWritten)?;
8397 if !probe.writable {
8398 return Err(GroupFailure::NotWritten(CoreError::TransportFailed {
8399 operation: "write".to_string(),
8400 node: "nodes.master".to_string(),
8401 detail: format!(
8402 "the {} collection cannot take the tag-group units ({})",
8403 spec.mode,
8404 probe.problems.join("; ")
8405 ),
8406 }));
8407 }
8408 let prepared = host
8409 .prepare(&spec, &tree)
8410 .await
8411 .map_err(GroupFailure::NotWritten)?;
8412 let lock = match HeldLock::acquire(deploy, ScanKind::Config, HOLD_S, &SystemClock).await {
8413 Ok(lock) => lock,
8414 Err(e) => {
8415 let _ = host.discard_interrupted(run_id).await;
8418 return Err(GroupFailure::NotWritten(e));
8419 }
8420 };
8421 let report = host
8424 .commit(&prepared)
8425 .await
8426 .map_err(GroupFailure::NotWritten)?;
8427 if report.stale {
8428 let _ = scan_and_wait(deploy, ScanKind::Config, group_scan_poll()).await;
8430 return Err(GroupFailure::Stale);
8431 }
8432 let restore = |why: String| restored(host, deploy, &spec.record, run_id, &lock, why);
8433 if let Some(failure) = &report.failure {
8434 return Err(restore(format!("the commit stopped: {failure}")).await);
8435 }
8436 match lock.check(deploy, LOCK_MARGIN_S, &SystemClock).await {
8437 Ok(LockVerdict::Held { .. }) => {}
8438 Ok(_) => return Err(restore("the scan-lock lapsed during the commit".to_string()).await),
8439 Err(e) => {
8440 return Err(restore(format!("the scan-lock could not be read ({})", e.code())).await);
8441 }
8442 }
8443 match scan_and_wait(deploy, ScanKind::Config, group_scan_poll()).await {
8444 Ok(ScanOutcome::Complete(_)) => {}
8445 Ok(ScanOutcome::NotComplete { .. }) => {
8446 return Err(restore("the config scan did not complete".to_string()).await);
8447 }
8448 Err(e) => return Err(restore(format!("the config scan failed ({})", e.code())).await),
8449 }
8450 let observed = match reader
8453 .read_tag_groups(&loaded.target.mode, &work.providers)
8454 .await
8455 {
8456 Ok(observed) => observed,
8457 Err(e) => {
8458 return Err(restore(format!("the units could not be read back ({})", e.code())).await);
8459 }
8460 };
8461 let mut differing: Vec<String> = spec
8462 .writes
8463 .iter()
8464 .filter(|w| observed.mode.get(&w.path).map(|b| b.sha256.as_str()) != Some(&w.sha256))
8465 .map(|w| w.path.as_str().to_string())
8466 .collect();
8467 differing.extend(
8468 spec.deletes
8469 .iter()
8470 .filter(|d| observed.mode.contains_key(&d.path))
8471 .map(|d| d.path.as_str().to_string()),
8472 );
8473 if !differing.is_empty() {
8474 return Err(restore(format!(
8475 "{} unit file(s) read back different: {}",
8476 differing.len(),
8477 differing.join(", ")
8478 ))
8479 .await);
8480 }
8481 let post = match host.fingerprint(&touched).await {
8482 Ok(post) => post,
8483 Err(e) => {
8484 return Err(GroupFailure::Unrestored(format!(
8485 "the units verified but their hashes could not be read ({}); the journal stays",
8486 e.code()
8487 )));
8488 }
8489 };
8490 let mut record = spec.record.clone();
8491 record.state = RecordState::Applied;
8492 record.post_fingerprint = Some(fingerprint(&post));
8493 host.finalize(&record).await.map_err(|e| {
8494 GroupFailure::Unrestored(format!(
8495 "the units verified but the run record could not be finalized ({}); the journal stays",
8496 e.code()
8497 ))
8498 })?;
8499 Ok((spec.writes.len(), spec.deletes.len(), record))
8500}
8501
8502async fn fresh_provider_settings(
8508 loaded: &LoadedProject,
8509 gateway: &dyn ProjectGateway,
8510 report: &TagsReport,
8511 run_dir: &RunDir,
8512 run_id: &RunId,
8513) -> Result<Vec<(ProviderSettingsPlan, Option<TagProviderRecord>)>, CoreError> {
8514 let mut out = Vec::new();
8515 for planned in &report.provider_settings {
8516 let name = planned.provider.as_str();
8517 let record = gateway.tag_provider_record(name).await?;
8518 let now =
8519 plan_provider_settings(name, provider_settings_of(loaded, name)?, record.as_ref());
8520 if &now != planned {
8521 let shown = |p: &ProviderSettingsPlan| {
8522 if p.properties.is_empty() {
8523 p.action.as_str().to_string()
8524 } else {
8525 format!("{} ({})", p.action.as_str(), p.properties.join(", "))
8526 }
8527 };
8528 return Err(CoreError::PlanStale {
8529 expected: shown(planned),
8530 actual: shown(&now),
8531 detail: format!("the settings of tag provider {name} changed since the plan"),
8532 });
8533 }
8534 if planned.action == ProviderSettingsAction::Modify
8535 && let Some(record) = &record
8536 {
8537 let bytes = serde_json::to_vec_pretty(record).map_err(|e| {
8538 CoreError::Internal(format!("a tag provider record did not serialize: {e}"))
8539 })?;
8540 let mut file = run_dir.create_private_file_nested(
8541 &["export", "tags"],
8542 &format!("{name}.provider.json"),
8543 )?;
8544 file.write_all(&bytes)
8545 .and_then(|()| file.sync_all())
8546 .map_err(|e| CoreError::DeployRunInvalid {
8547 run_id: run_id.as_str().to_string(),
8548 problem: format!(
8549 "could not write the settings pre-image of tag provider {name} ({:?})",
8550 e.kind()
8551 ),
8552 })?;
8553 }
8554 out.push((planned.clone(), record));
8555 }
8556 Ok(out)
8557}
8558
8559fn note_tag_step(result: &mut DeployProjectsResult, step: &TagStep) {
8562 let Some(tags) = result.tags.as_mut() else {
8563 return;
8564 };
8565 tags.carried_count = step.carried.len();
8566 tags.carried = step.carried.iter().take(LIST_CAP).cloned().collect();
8567 tags.verify = step.verify.clone();
8568 tags.deleted_count = step.deleted.len();
8569 tags.deleted = step.deleted.iter().take(LIST_CAP).cloned().collect();
8570 if !step.absent.is_empty() {
8573 tags.already_absent_count += step.absent.len();
8574 tags.already_absent.extend(step.absent.iter().cloned());
8575 tags.already_absent
8576 .sort_by(|a, b| a.as_bytes().cmp(b.as_bytes()));
8577 tags.already_absent.dedup();
8578 tags.already_absent.truncate(LIST_CAP);
8579 }
8580 if step.verify.as_ref().is_some_and(|v| !v.settled) {
8581 result.warnings.push(format!(
8582 "the owned tag providers did not read the same twice within the verify bound; \
8583 the last read was compared and equals the repo ({} read(s))",
8584 step.verify.as_ref().map_or(0, |v| v.reads)
8585 ));
8586 }
8587}
8588
8589async fn write_tag_records(
8598 result: &mut DeployProjectsResult,
8599 loaded: &LoadedProject,
8600 run_id: &RunId,
8601 writer: Option<&dyn TagRecordWriter>,
8602 events: &mut RunEvents<'_>,
8603 changed_gateway: bool,
8604) -> Result<(), CoreError> {
8605 let (Some(tags), Some(writer)) = (result.tags.as_ref(), writer) else {
8606 return Ok(());
8607 };
8608 let tags_sha256 = tags.tags_sha256.clone();
8609 for provider in sorted(&loaded.owned_tags) {
8610 let mut groups: Vec<String> = provider
8611 .groups
8612 .as_ref()
8613 .and_then(|g| g.get("tagGroups"))
8614 .and_then(Value::as_array)
8615 .into_iter()
8616 .flatten()
8617 .filter_map(|g| g.get("name").and_then(Value::as_str))
8618 .map(str::to_string)
8619 .collect();
8620 groups.sort_by(|a, b| a.as_bytes().cmp(b.as_bytes()));
8621 groups.dedup();
8622 let record = TagRecord {
8623 schema_version: 1,
8624 gateway: result.gateway.clone(),
8625 provider: provider.name.clone(),
8626 project: result.project.clone(),
8627 release: loaded.release.tag.clone(),
8628 run_id: run_id.as_str().to_string(),
8629 deployed_at: rfc3339_now_utc(),
8630 tags_sha256: tags_sha256.clone(),
8631 nodes: provider.nodes.keys().cloned().collect(),
8632 groups,
8633 };
8634 match writer
8635 .write_tag_record(&result.gateway, &provider.name, &record.to_bytes())
8636 .await
8637 {
8638 Ok(()) => events.emit_late(ProjectsEvent::TagRecordWritten {
8639 provider: provider.name.clone(),
8640 }),
8641 Err(err) => {
8642 events.emit_late(ProjectsEvent::TagRecordFailed {
8643 provider: provider.name.clone(),
8644 code: err.code().to_string(),
8645 });
8646 if !changed_gateway {
8647 return Err(err);
8648 }
8649 result.warnings.push(format!(
8650 "the run is verified but the tag record of provider {} could not be written ({})",
8651 provider.name,
8652 err.code()
8653 ));
8654 }
8655 }
8656 }
8657 Ok(())
8658}
8659
8660fn sorted(providers: &[NativeProvider]) -> Vec<&NativeProvider> {
8662 let mut out: Vec<&NativeProvider> = providers.iter().collect();
8663 out.sort_by(|a, b| a.name.as_bytes().cmp(b.name.as_bytes()));
8664 out
8665}
8666
8667fn import_outcome_unknown(err: &CoreError) -> bool {
8674 match err {
8675 CoreError::Network {
8676 source: Some(source),
8677 ..
8678 } => !(source.is_connect() || source.is_builder()),
8679 CoreError::Network { source: None, .. } | CoreError::Internal(_) => true,
8680 _ => false,
8681 }
8682}
8683
8684fn finish_recorded(
8692 result: &mut DeployProjectsResult,
8693 events: &mut RunEvents<'_>,
8694 backup_unsynced: bool,
8695 record_written: bool,
8696) {
8697 result.record_written = record_written;
8698 result.outcome = if result.tags.is_some() {
8699 ProjectOutcome::Applied
8700 } else {
8701 ProjectOutcome::Recorded
8702 };
8703 if backup_unsynced && result.tags.is_some() {
8704 mark_backup_unsynced(result);
8705 }
8706 if record_written {
8707 events.emit_late(ProjectsEvent::RecordWritten);
8708 }
8709 events.emit_late(ProjectsEvent::Finished {
8710 outcome: result.outcome,
8711 });
8712 note_late_log_failure(result, events);
8713}
8714
8715fn mark_backup_unsynced(result: &mut DeployProjectsResult) {
8718 result.outcome = ProjectOutcome::AppliedBackupUnsynced;
8719 result.ok = false;
8720 result.slug = Some(BACKUP_UNSYNCED_SLUG.to_string());
8721}
8722
8723fn backup_event(event: TagsBackupEvent) -> ProjectsEvent {
8725 match event {
8726 TagsBackupEvent::WaitStarted { budget_s } => ProjectsEvent::BackupWaitStarted { budget_s },
8727 TagsBackupEvent::Resync { ok, code } => ProjectsEvent::BackupResync { ok, code },
8728 TagsBackupEvent::Verdict { converged, reason } => ProjectsEvent::BackupVerdict {
8729 converged,
8730 reason: reason.map(|reason| reason.as_str()),
8731 },
8732 }
8733}
8734
8735async fn backup_step(
8742 result: &mut DeployProjectsResult,
8743 providers: &[&str],
8744 surfaces: &ProjectsSurfaces<'_>,
8745 opts: &ProjectsRunOptions<'_>,
8746 since_ms: i64,
8747 events: &mut RunEvents<'_>,
8748) -> Option<bool> {
8749 let backup = opts.backup?;
8750 result.tags.as_ref()?;
8751 let mut master = BTreeMap::new();
8752 let mut unread = None;
8753 for name in providers {
8754 match surfaces.gateway.tag_export(name).await {
8755 Ok(bytes) => match serde_json::from_slice::<Value>(&bytes) {
8756 Ok(tree) => {
8757 master.insert((*name).to_string(), tree);
8758 }
8759 Err(_) => {
8760 unread = Some(format!("[{name}] is not JSON"));
8761 break;
8762 }
8763 },
8764 Err(err) => {
8765 unread = Some(format!("[{name}] {}", err.code()));
8766 break;
8767 }
8768 }
8769 }
8770 let report = match unread {
8771 None => {
8772 let resync = surfaces.deploy.map(DeployResync);
8773 converge_tags(
8774 &master,
8775 backup,
8776 resync.as_ref().map(|r| r as &dyn MasterResync),
8777 opts.backup_policy,
8778 since_ms,
8779 &mut |event| events.emit_late(backup_event(event)),
8780 )
8781 .await
8782 }
8783 Some(why) => {
8784 result.warnings.push(format!(
8785 "the backup check could not read the master's export ({why}), so the backup's \
8786 copy of the owned tags is not proven"
8787 ));
8788 let mut names: Vec<String> = providers.iter().map(|p| (*p).to_string()).collect();
8789 names.sort_by(|a, b| a.as_bytes().cmp(b.as_bytes()));
8790 TagsBackupReport {
8791 converged: false,
8792 reason: Some(crate::deploy::pair::UnsyncedReason::Unreachable),
8793 budget_s: opts.backup_policy.budget.as_secs(),
8794 elapsed_s: 0,
8795 resync_sent: false,
8796 providers: names,
8797 differing: Vec::new(),
8798 }
8799 }
8800 };
8801 let converged = report.converged;
8802 if let Some(tags) = result.tags.as_mut() {
8803 tags.backup = Some(report);
8804 }
8805 Some(converged)
8806}
8807
8808fn note_late_log_failure(result: &mut DeployProjectsResult, events: &RunEvents<'_>) {
8810 if events.late_failure {
8811 result
8812 .warnings
8813 .push("the run's events log could not be written to the end".to_string());
8814 }
8815}
8816
8817#[cfg(test)]
8820pub(crate) fn sample_result() -> DeployProjectsResult {
8821 DeployProjectsResult {
8822 env: "prod".to_string(),
8823 gateway: "gw".to_string(),
8824 project: "CapApp".to_string(),
8825 release: "v1.0.0".to_string(),
8826 check: false,
8827 action: ProjectAction::Record,
8828 would_change: true,
8829 artifact_sha256: "a".repeat(64),
8830 live_sha256: Some("a".repeat(64)),
8831 members: MemberDelta::default(),
8832 removed: Vec::new(),
8833 removed_count: 0,
8834 excluded: Vec::new(),
8835 excluded_count: 0,
8836 unmanaged: Vec::new(),
8837 unmanaged_count: 0,
8838 parent: None,
8839 previous_release: None,
8840 recorded_sha256: None,
8841 live_matches_last_deploy: None,
8842 allow_downgrade: false,
8843 allow_first_change: None,
8844 warnings: Vec::new(),
8845 run_id: None,
8846 run_dir: None,
8847 record_written: false,
8848 restore_command: None,
8849 children: Vec::new(),
8850 missing_tags: Some(Vec::new()),
8851 missing_tags_count: Some(0),
8852 missing_tags_unknown_providers: Some(Vec::new()),
8853 missing_tags_unknown_providers_count: Some(0),
8854 missing_tags_unavailable: None,
8855 missing_tags_limit: MISSING_TAGS_LIMIT,
8856 tags: None,
8857 outcome: ProjectOutcome::Checked,
8858 ok: true,
8859 slug: None,
8860 }
8861}
8862
8863#[cfg(test)]
8864mod tests {
8865 use super::*;
8866
8867 fn sample_zip() -> Vec<u8> {
8868 use std::io::Write as _;
8869 let mut writer = zip::ZipWriter::new(std::io::Cursor::new(Vec::new()));
8870 let options = zip::write::SimpleFileOptions::default()
8871 .compression_method(zip::CompressionMethod::Deflated);
8872 for (name, bytes) in [
8873 ("project.json", br#"{"title":"T"}"#.as_slice()),
8874 ("ignition/s/resource.json", br#"{"files":["a.py"]}"#),
8875 ("ignition/s/a.py", &[b'x'; 4096]),
8876 ] {
8877 writer.start_file(name, options).expect("start");
8878 writer.write_all(bytes).expect("write");
8879 }
8880 writer.finish().expect("finish").into_inner()
8881 }
8882
8883 #[test]
8884 fn check_artifact_zip_refuses_truncated() {
8885 let whole = sample_zip();
8886 check_artifact_zip(&whole).expect("the whole zip passes");
8887 let half = &whole[..whole.len() / 2];
8888 let err = check_artifact_zip(half).expect_err("half a zip");
8889 assert_eq!(err.code(), "invalid_import_file");
8890 let err = check_artifact_zip(&whole[..whole.len() - 40]).expect_err("tail cut");
8892 assert_eq!(err.code(), "invalid_import_file");
8893 let err = check_artifact_zip(b"not a zip").expect_err("no magic");
8894 assert_eq!(err.code(), "invalid_import_file");
8895 }
8896
8897 use super::sample_result;
8898
8899 #[test]
8903 fn a_log_failure_after_the_record_is_written_is_a_warning() {
8904 let dir = tempfile::tempdir().expect("tempdir");
8905 let path = dir.path().join("events.ndjson");
8906 std::fs::write(&path, b"").expect("seed");
8907 let read_only = std::fs::File::open(&path).expect("open read-only");
8909 let mut events = RunEvents {
8910 log: EventLog::from_file(read_only),
8911 observer: &NoProjectsObserver,
8912 late_failure: false,
8913 };
8914 let mut result = sample_result();
8915
8916 finish_recorded(&mut result, &mut events, false, true);
8917
8918 assert!(result.record_written, "the record is on the host");
8919 assert_eq!(result.outcome, ProjectOutcome::Recorded);
8920 assert!(result.ok, "a log failure does not fail the run");
8921 assert!(events.late_failure);
8922 assert!(
8923 result
8924 .warnings
8925 .iter()
8926 .any(|w| w.contains("events log could not be written")),
8927 "{:?}",
8928 result.warnings
8929 );
8930 }
8931
8932 #[test]
8933 fn a_built_artifact_passes_the_check() {
8934 let dir = tempfile::tempdir().expect("tempdir");
8935 for (rel, bytes) in [
8936 ("project.json", br#"{"title":"T"}"#.as_slice()),
8937 ("ignition/s/resource.json", br#"{"files":["a.py"]}"#),
8938 ("ignition/s/a.py", b"pass"),
8939 ] {
8940 let path = dir.path().join(rel);
8941 std::fs::create_dir_all(path.parent().expect("parent")).expect("mkdir");
8942 std::fs::write(path, bytes).expect("write");
8943 }
8944 let artifact = build_artifact(dir.path()).expect("artifact");
8945 check_artifact_zip(&artifact.zip).expect("a built artifact is a valid import");
8946 }
8947}