use std::collections::{BTreeMap, BTreeSet, VecDeque};
use std::io::Write as _;
use std::path::PathBuf;
use std::time::{Duration, Instant};
use async_trait::async_trait;
use serde::Serialize;
use serde_json::Value;
use super::deploy::load_target;
use super::projects::validate_import;
use super::resources::export_zip_bytes;
use super::workspace::rfc3339_now_utc;
use crate::client::deploy_api::scan_and_wait;
use crate::client::projects::{ImportOutcome, ProjectRecord};
use crate::client::query::ListQuery;
use crate::client::scan::{LockVerdict, ScanKind, ScanOutcome};
use crate::client::tags::{
CollisionPolicy, TagImportAnswer, TagProviderCreate, TagProviderModify, TagProviderRecord,
};
use crate::client::{DeployApi, GatewayApi};
use crate::deploy::apply::fingerprint::fingerprint;
use crate::deploy::apply::hold::SystemClock;
use crate::deploy::apply::hold::{HOLD_S, HeldLock, LOCK_MARGIN_S};
use crate::deploy::apply::host::{ApplyHost, BatchSpec, DeleteOp, WriteOp};
use crate::deploy::apply::journal::{Journal, JournalPhase};
use crate::deploy::apply::preflight::{
DesignerSeen, DesignersPolicy, DesignersVerdict, designers_detail, designers_gate,
read_designers, wrong_target_failure, wrong_target_gate,
};
use crate::deploy::apply::record::{RecordState, RunRecord, TouchedEntry};
use crate::deploy::apply::rollback::live_changes;
use crate::deploy::apply::run_store::{EventLog, RunDir, RunId, run_dir_rel};
use crate::deploy::apply::{BACKUP_UNSYNCED_SLUG, ROLLED_BACK_SLUG};
use crate::deploy::exclude;
use crate::deploy::ledger::{
LedgerRoot, journal_location, pre_image_location, project_journal_location,
release_record_location, tag_record_location,
};
use crate::deploy::lint::{Sends, scan_tag_node};
use crate::deploy::manifest::Manifest;
use crate::deploy::native_tags::{
NativeNode, NativeProvider, NonCanonical, TAG_GROUPS_FILE, check_canonical_form,
read_provider_with_raw,
};
use crate::deploy::pair::{BackupPolicy, wall_clock_ms};
use crate::deploy::project_artifact::{
Artifact, CanonicalDigest, build_artifact, canonical_digest,
};
use crate::deploy::project_run::{
CreatedProof, JournalState, PreImageFile, PreImageProvider, PreImageRun, ProjectJournal,
};
use crate::deploy::project_tags::{
FILES_PER_ENTRY, Lookup, MISSING_TAGS_LIMIT, MissingTag, TagRef, TagRefs, TagTree,
TagsUnavailable, UnknownProvider, scan_artifact,
};
use crate::deploy::projects::{
MemberDelta, PROJECT_FILE_REL, ParentLink, ProjectAction, ProjectFile, ProjectMeta,
ProviderSettings, ReadOnlySources, Release, ReleaseOrder, ReleaseRecord, check_parent,
check_tag_ownership, gateway_manifest_rel, load_project_file, load_provider_settings,
member_delta, parent_chain, read_project_meta,
};
use crate::deploy::tag_apply::{
AppliedSteps, Carried, ImportBatch, REMOVED_PROPERTY_POLICY, TAG_DELETE_DESCENDANTS, carried,
created_deletes_now, delete_calls, import_batches, rollback_over_cap, rollback_plan,
uncarriable_in_batch,
};
use crate::deploy::tag_canon::{CanonOpts, canonical, diff_properties, flatten_export, to_node};
use crate::deploy::tag_converge::{
BackupTags, DeployResync, MasterResync, TagsBackupEvent, TagsBackupReport, converge_tags,
};
use crate::deploy::tag_delete::{RouteHandshake, TagDeleteRoute, delete_chunks, wait_ready};
use crate::deploy::tag_groups::{
GroupPlan, GroupSummary, LiveGroups, check_group_refs, live_group_users, parse_group_value,
plan_groups, referenced_groups,
};
use crate::deploy::tag_plan::{
NodeChange, ProviderPlan, ProviderSettingsAction, ProviderSettingsPlan, TagDeleteRouteReport,
TagJournal, TagPlan, TagRollback, TagVerify, TagsReport, plan_provider_settings, plan_tags,
removed_types_in_use, repo_tags_sha256,
};
use crate::deploy::tag_record::TagRecord;
use crate::deploy::target::{NodeTarget, Target};
use crate::deploy::transport::projects::{
ProjectHost, ProjectRunWriter, ReleaseRecordWriter, TagRecordWriter, git_entry_location,
project_run_writer_for_node, release_writer_for_node, tag_record_names_problem,
tag_record_writer_for_node,
};
use crate::deploy::transport::remote::safe_remote_path;
use crate::deploy::transport::{TreeReader, apply_host_for_node, for_node};
use crate::deploy::tree::read_repo_file;
use crate::deploy::tree::{RenderedTree, sha256_hex};
use crate::error::{
APPLY_LIST_SHOWN, CoreError, DeployFinding, PreflightFailure, RollbackFailure, RollbackScope,
};
pub const LIST_CAP: usize = 200;
pub const CHILDREN_SETTLE_S: u64 = 30;
pub const CHILDREN_POLL_S: u64 = 5;
pub const TAG_VERIFY_SETTLE_S: u64 = 30;
pub const TAG_VERIFY_POLL_S: u64 = 5;
pub const PROJECTS_CANARY_EXEMPT: &[&str] = &["export"];
#[async_trait]
pub trait ProjectGateway: Send + Sync {
async fn projects(&self) -> Result<Vec<ProjectRecord>, CoreError>;
async fn project_find(&self, name: &str) -> Result<Option<ProjectRecord>, CoreError>;
async fn export_bytes(&self, name: &str) -> Result<Vec<u8>, CoreError>;
async fn designers(&self) -> Result<Vec<DesignerSeen>, CoreError>;
async fn gateway_name(&self) -> Result<Option<String>, CoreError>;
async fn tag_provider_names(&self) -> Result<Vec<String>, CoreError>;
async fn tag_export(&self, provider: &str) -> Result<Vec<u8>, CoreError>;
async fn project_import(
&self,
name: &str,
zip: Vec<u8>,
overwrite: bool,
) -> Result<ImportOutcome, CoreError>;
async fn tag_import(
&self,
provider: &str,
path: &str,
policy: CollisionPolicy,
body: Vec<u8>,
) -> Result<TagImportAnswer, CoreError>;
async fn tag_provider_record(&self, name: &str)
-> Result<Option<TagProviderRecord>, CoreError>;
async fn tag_provider_create(&self, body: &[TagProviderCreate]) -> Result<(), CoreError>;
async fn tag_provider_modify(&self, body: &[TagProviderModify]) -> Result<(), CoreError>;
async fn tag_provider_delete(&self, name: &str, signature: &str) -> Result<(), CoreError>;
}
pub struct GatewayProjects<'a>(pub &'a dyn GatewayApi);
#[async_trait]
impl ProjectGateway for GatewayProjects<'_> {
async fn projects(&self) -> Result<Vec<ProjectRecord>, CoreError> {
Ok(self.0.projects(&ListQuery::default()).await?.items)
}
async fn project_find(&self, name: &str) -> Result<Option<ProjectRecord>, CoreError> {
match self.0.project_find(name).await {
Ok(record) => Ok(Some(record)),
Err(CoreError::NotFound { .. }) => Ok(None),
Err(err) => Err(err),
}
}
async fn export_bytes(&self, name: &str) -> Result<Vec<u8>, CoreError> {
export_zip_bytes(self.0, name).await
}
async fn designers(&self) -> Result<Vec<DesignerSeen>, CoreError> {
read_designers(self.0).await
}
async fn gateway_name(&self) -> Result<Option<String>, CoreError> {
Ok(self
.0
.gateway_info()
.await?
.name
.filter(|name| !name.trim().is_empty()))
}
async fn tag_provider_names(&self) -> Result<Vec<String>, CoreError> {
Ok(self
.0
.tag_provider_list(&ListQuery::default())
.await?
.items
.into_iter()
.map(|record| record.name)
.collect())
}
async fn tag_export(&self, provider: &str) -> Result<Vec<u8>, CoreError> {
self.0.tags_export_provider(provider).await
}
async fn project_import(
&self,
name: &str,
zip: Vec<u8>,
overwrite: bool,
) -> Result<ImportOutcome, CoreError> {
self.0.project_import(name, zip, overwrite).await
}
async fn tag_import(
&self,
provider: &str,
path: &str,
policy: CollisionPolicy,
body: Vec<u8>,
) -> Result<TagImportAnswer, CoreError> {
self.0.tags_import(provider, path, policy, body).await
}
async fn tag_provider_record(
&self,
name: &str,
) -> Result<Option<TagProviderRecord>, CoreError> {
match self.0.tag_provider_find(name).await {
Ok(record) => Ok(Some(record)),
Err(CoreError::NotFound { .. }) => Ok(None),
Err(err) => Err(err),
}
}
async fn tag_provider_create(&self, body: &[TagProviderCreate]) -> Result<(), CoreError> {
self.0.tag_provider_create(body).await
}
async fn tag_provider_modify(&self, body: &[TagProviderModify]) -> Result<(), CoreError> {
self.0.tag_provider_modify(body).await
}
async fn tag_provider_delete(&self, name: &str, signature: &str) -> Result<(), CoreError> {
self.0.tag_provider_delete(name, signature).await
}
}
pub struct ProjectsSurfaces<'a> {
pub gateway: &'a dyn ProjectGateway,
pub host: &'a dyn ProjectHost,
pub tree: Option<&'a dyn TreeReader>,
pub apply_host: Option<&'a dyn ApplyHost>,
pub deploy: Option<&'a dyn DeployApi>,
pub tag_delete: TagDeleteSurface<'a>,
}
#[derive(Clone)]
pub enum TagDeleteSurface<'a> {
Configured(&'a dyn TagDeleteRoute),
NotConfigured {
missing: Vec<String>,
},
}
struct RouteState<'a> {
route: Option<&'a dyn TagDeleteRoute>,
missing: Vec<String>,
handshake: Option<RouteHandshake>,
problem: Option<String>,
}
impl<'a> RouteState<'a> {
fn of(
surface: &TagDeleteSurface<'a>,
answer: Option<Result<RouteHandshake, CoreError>>,
) -> Self {
match surface {
TagDeleteSurface::Configured(route) => {
let (handshake, problem) = match answer {
Some(Ok(handshake)) => (Some(handshake), None),
Some(Err(err)) => (None, Some(err.code().to_string())),
None => (None, None),
};
Self {
route: Some(*route),
missing: Vec::new(),
handshake,
problem,
}
}
TagDeleteSurface::NotConfigured { missing } => Self {
route: None,
missing: missing.clone(),
handshake: None,
problem: None,
},
}
}
fn ready(&self) -> Option<(&'a dyn TagDeleteRoute, &RouteHandshake)> {
Some((self.route?, self.handshake.as_ref()?))
}
fn report(&self, owned: &[&str]) -> TagDeleteRouteReport {
let mut providers_not_allowed: Vec<String> = match &self.handshake {
Some(handshake) => owned
.iter()
.filter(|name| !handshake.providers.iter().any(|p| p == *name))
.map(|name| (*name).to_string())
.collect(),
None => Vec::new(),
};
providers_not_allowed.sort_by(|a, b| a.as_bytes().cmp(b.as_bytes()));
TagDeleteRouteReport {
configured: self.route.is_some(),
missing: self.missing.clone(),
ready: self.handshake.is_some(),
route_version: self.handshake.as_ref().map(|h| h.route_version.clone()),
providers_not_allowed,
problem: self.problem.clone(),
}
}
fn kept_reason(&self, provider: &str) -> String {
match (self.route, &self.handshake) {
(None, _) => "created by this run; this release has no tag delete".to_string(),
(Some(_), None) => format!(
"created by this run; the tag-delete route is not ready ({}), so it was not \
removed",
self.problem.as_deref().unwrap_or("no handshake")
),
(Some(_), Some(handshake)) if !handshake.providers.iter().any(|p| p == provider) => {
format!(
"created by this run; the tag-delete route does not allow provider \
{provider}, so it was not removed"
)
}
(Some(_), Some(_)) => "created by this run; it holds a node this run did not \
create, so it was not removed"
.to_string(),
}
}
}
fn owned_names(loaded: &LoadedProject) -> Vec<&str> {
sorted(&loaded.owned_tags)
.into_iter()
.map(|p| p.name.as_str())
.collect()
}
#[derive(Debug, Clone)]
pub struct ProjectsInput {
pub repo: PathBuf,
pub targets: PathBuf,
pub env: String,
pub release: String,
pub allow_downgrade: bool,
pub allow_first_change: Option<String>,
pub allow_destructive: bool,
pub artifact_sha256: Option<String>,
pub tags_sha256: Option<String>,
}
pub const FIRST_CHANGE_REASON_MAX: usize = 500;
#[derive(Debug)]
pub struct LoadedProject {
pub file: ProjectFile,
pub meta: ProjectMeta,
pub target_file: String,
pub target: Target,
pub artifact: Artifact,
pub release: Release,
pub owned_tags: Vec<NativeProvider>,
pub provider_settings: Vec<ProviderSettings>,
pub tag_exclusions: Vec<String>,
pub ownership_warnings: Vec<String>,
pub non_canonical: std::collections::BTreeMap<String, Vec<NonCanonical>>,
}
pub fn load_project(input: &ProjectsInput) -> Result<LoadedProject, CoreError> {
let release = Release::parse(&input.release)?;
load_with(input, release)
}
pub fn load_project_for_rollback(input: &ProjectsInput) -> Result<LoadedProject, CoreError> {
load_with(
input,
Release {
tag: String::new(),
version: semver::Version::new(0, 0, 0),
},
)
}
fn load_with(input: &ProjectsInput, release: Release) -> Result<LoadedProject, CoreError> {
if let Some(reason) = &input.allow_first_change {
let chars = reason.chars().count();
if chars == 0 || chars > FIRST_CHANGE_REASON_MAX {
return Err(CoreError::InvalidInput {
reason: format!(
"--allow-first-change <reason>: the reason is 1 to {FIRST_CHANGE_REASON_MAX} characters"
),
});
}
}
if let Some(hash) = &input.artifact_sha256
&& !is_lower_hex64(hash)
{
return Err(CoreError::InvalidInput {
reason: "--artifact-sha256 must be 64 lowercase hex characters (the \
`artifact_sha256` of `ign deploy projects --check --json`)"
.to_string(),
});
}
if let Some(hash) = &input.tags_sha256
&& !is_lower_hex64(hash)
{
return Err(CoreError::InvalidInput {
reason: "--tags-sha256 must be 64 lowercase hex characters (the \
`data.tags.tags_sha256` of `ign deploy projects --check --json`)"
.to_string(),
});
}
let file = load_project_file(&input.repo)?;
file.require_env(&input.env)?;
let meta = read_project_meta(&input.repo)?;
let (target_file, target) = load_target(&input.targets, &input.env)?;
let (tag_exclusions, ownership_warnings) = if file.tag_providers.is_empty() {
(exclude::effective(&target, None), Vec::new())
} else {
let manifest_rel = gateway_manifest_rel(&target.gateway);
let manifest = read_gateway_manifest(&input.targets, &manifest_rel)?;
let manifest = manifest.as_ref().map(|m| (manifest_rel.as_str(), m));
let sources = ReadOnlySources::new(&target_file, &target, manifest);
let warnings = check_tag_ownership(&file, &sources, manifest)?;
(
exclude::effective(&target, manifest.map(|(_, m)| m)),
warnings,
)
};
record_names_fit_the_host(
&file.project,
&file.tag_providers,
&target.gateway,
&target_file,
&target.nodes.master,
)?;
let artifact = build_artifact(&input.repo)?;
check_artifact_zip(&artifact.zip)?;
if let Some(expected) = &input.artifact_sha256
&& *expected != artifact.digest.sha256
{
return Err(CoreError::ProjectHashMismatch {
project: file.project.clone(),
hash: "artifact_sha256",
expected: expected.clone(),
actual: artifact.digest.sha256.clone(),
});
}
let provider_settings = file
.tag_providers
.iter()
.map(|provider| load_provider_settings(&input.repo, provider))
.collect::<Result<Vec<_>, _>>()?;
let mut owned_tags = Vec::new();
let mut non_canonical = std::collections::BTreeMap::new();
for provider in &file.tag_providers {
let (tags, raw) = read_provider_with_raw(&input.repo, provider)?;
if tags.nodes.is_empty() {
return Err(CoreError::deploy_tree_invalid(
format!("{}/{provider}", crate::deploy::native_tags::TAGS_DIR),
None,
None,
"holds no tag node: an owned provider is never deployed empty; drop it from \
tag_providers in deploy/project.yaml or keep at least one node"
.to_string(),
));
}
non_canonical.insert(provider.clone(), check_canonical_form(&tags, &raw));
owned_tags.push(tags);
}
if let Some(expected) = &input.tags_sha256 {
if owned_tags.is_empty() {
return Err(CoreError::InvalidInput {
reason: "--tags-sha256 is given, but deploy/project.yaml lists no \
tag_providers: there is no owned tag content to compare it with"
.to_string(),
});
}
let actual = repo_tags_sha256(&owned_tags);
if *expected != actual {
return Err(CoreError::ProjectHashMismatch {
project: file.project.clone(),
hash: "tags_sha256",
expected: expected.clone(),
actual,
});
}
}
Ok(LoadedProject {
file,
meta,
target_file,
target,
artifact,
release,
owned_tags,
provider_settings,
tag_exclusions,
ownership_warnings,
non_canonical,
})
}
fn read_gateway_manifest(
targets: &std::path::Path,
rel: &str,
) -> Result<Option<Manifest>, CoreError> {
match std::fs::symlink_metadata(targets.join(rel)) {
Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
_ => {}
}
let bytes = read_repo_file(targets, rel, None)?;
Manifest::parse(rel, &bytes).map(Some)
}
fn record_names_fit_the_host(
project: &str,
tag_providers: &[String],
gateway: &str,
target_file: &str,
master: &NodeTarget,
) -> Result<(), CoreError> {
if master.ssh.is_some() {
if let Err(why) = safe_remote_path(project) {
return Err(CoreError::deploy_tree_invalid(
PROJECT_FILE_REL,
None,
None,
format!(
"project: the name cannot be used for the release record on an ssh node ({why})"
),
));
}
if let Err(why) = safe_remote_path(gateway) {
return Err(CoreError::deploy_tree_invalid(
target_file,
None,
None,
format!(
"gateway: the name cannot be used for the release record on an ssh node ({why})"
),
));
}
}
let findings: Vec<crate::error::DeployFinding> = tag_providers
.iter()
.enumerate()
.filter_map(|(i, provider)| {
tag_record_names_problem(gateway, provider, master).map(|why| {
crate::error::DeployFinding {
file: PROJECT_FILE_REL.to_string(),
line: None,
column: None,
message: format!(
"tag_providers[{i}]: the name cannot be used for the tag record on \
the master node ({why})"
),
}
})
})
.collect();
if !findings.is_empty() {
return Err(CoreError::deploy_findings(findings));
}
Ok(())
}
fn is_lower_hex64(text: &str) -> bool {
text.len() == 64
&& text
.bytes()
.all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b))
}
pub(crate) fn check_artifact_zip(zip: &[u8]) -> Result<(), CoreError> {
validate_import(zip)
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum ProjectOutcome {
Checked,
Skipped,
Created,
Imported,
Recorded,
ImportedUnverified,
Applied,
TagsRolledBack,
RolledBack,
AppliedBackupUnsynced,
}
impl ProjectOutcome {
pub fn as_str(self) -> &'static str {
match self {
ProjectOutcome::Checked => "checked",
ProjectOutcome::Skipped => "skipped",
ProjectOutcome::Created => "created",
ProjectOutcome::Imported => "imported",
ProjectOutcome::Recorded => "recorded",
ProjectOutcome::ImportedUnverified => "imported_unverified",
ProjectOutcome::Applied => "applied",
ProjectOutcome::TagsRolledBack => "tags_rolled_back",
ProjectOutcome::RolledBack => "rolled_back",
ProjectOutcome::AppliedBackupUnsynced => "applied_backup_unsynced",
}
}
}
pub const PROJECT_UNVERIFIED_SLUG: &str = "deploy_project_unverified";
#[derive(Debug, Clone, Serialize)]
pub struct DeployProjectsResult {
pub env: String,
pub gateway: String,
pub project: String,
pub release: String,
pub check: bool,
pub action: ProjectAction,
pub would_change: bool,
pub artifact_sha256: String,
pub live_sha256: Option<String>,
pub members: MemberDelta,
pub removed: Vec<String>,
pub removed_count: usize,
pub excluded: Vec<String>,
pub excluded_count: usize,
pub unmanaged: Vec<String>,
pub unmanaged_count: usize,
pub parent: Option<String>,
pub previous_release: Option<String>,
pub recorded_sha256: Option<String>,
pub live_matches_last_deploy: Option<bool>,
pub allow_downgrade: bool,
pub allow_first_change: Option<String>,
pub warnings: Vec<String>,
pub run_id: Option<String>,
pub run_dir: Option<String>,
pub record_written: bool,
pub restore_command: Option<String>,
pub children: Vec<ChildReport>,
pub missing_tags: Option<Vec<MissingTag>>,
pub missing_tags_count: Option<usize>,
pub missing_tags_unknown_providers: Option<Vec<UnknownProvider>>,
pub missing_tags_unknown_providers_count: Option<usize>,
pub missing_tags_unavailable: Option<Vec<TagsUnavailable>>,
pub missing_tags_limit: &'static str,
pub tags: Option<Box<TagsReport>>,
pub outcome: ProjectOutcome,
pub ok: bool,
pub slug: Option<String>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum ChildStatus {
Ok,
Mismatch,
Unverified,
}
impl ChildStatus {
pub fn as_str(self) -> &'static str {
match self {
ChildStatus::Ok => "ok",
ChildStatus::Mismatch => "mismatch",
ChildStatus::Unverified => "unverified",
}
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct ChildReport {
pub name: String,
pub status: ChildStatus,
pub release: Option<String>,
pub recorded_sha256: Option<String>,
pub live_sha256: Option<String>,
}
impl DeployProjectsResult {
pub fn failure_exit_code(&self) -> Option<u8> {
matches!(
self.outcome,
ProjectOutcome::ImportedUnverified
| ProjectOutcome::TagsRolledBack
| ProjectOutcome::RolledBack
| ProjectOutcome::AppliedBackupUnsynced
)
.then_some(6)
}
}
fn shell_word(text: &str) -> String {
let plain = !text.is_empty()
&& text.chars().all(|c| {
c.is_ascii_alphanumeric() || matches!(c, '_' | '-' | '.' | '/' | ':' | '@' | '+' | ',')
});
if plain {
text.to_string()
} else {
format!("'{}'", text.replace('\'', "'\\''"))
}
}
fn restore_command(project: &str, run_dir_rel: &str) -> String {
format!(
"ign project import {} --file {} --collision-policy overwrite --yes",
shell_word(project),
shell_word(&format!("{run_dir_rel}/export/{project}.zip")),
)
}
fn capped(mut items: Vec<String>) -> (Vec<String>, usize) {
let count = items.len();
items.truncate(LIST_CAP);
(items, count)
}
fn cap_list<T>(mut items: Vec<T>) -> (Vec<T>, usize) {
let count = items.len();
items.truncate(LIST_CAP);
(items, count)
}
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct TagCheck {
pub missing: Option<Vec<MissingTag>>,
pub unknown_providers: Option<Vec<UnknownProvider>>,
pub unavailable: Vec<TagsUnavailable>,
}
fn unavailable_reason(what: &str, err: &CoreError) -> String {
format!("{what} ({})", err.code())
}
const UNRESOLVED_PATHS_SHOWN: usize = 3;
fn unresolved_reason(paths: &std::collections::BTreeSet<String>) -> String {
let shown: Vec<&str> = paths
.iter()
.take(UNRESOLVED_PATHS_SHOWN)
.map(String::as_str)
.collect();
let rest = paths.len() - shown.len();
let mut reason = format!(
"the provider's tag export does not resolve the UDT type of {} tag path(s): {}",
paths.len(),
shown.join(", ")
);
if rest > 0 {
reason.push_str(&format!(" and {rest} more"));
}
reason
}
fn strip_property(path: &str) -> Option<&str> {
let start = path.rfind('/').map_or(0, |at| at + 1);
let dot = path[start..].find('.')?;
(dot > 0).then(|| &path[..start + dot])
}
fn provider_shaped(name: &str) -> bool {
name.chars().any(char::is_alphabetic)
&& name
.chars()
.all(|c| c.is_alphanumeric() || matches!(c, '_' | '-' | ' '))
&& !name.starts_with(' ')
&& !name.ends_with(' ')
&& !name.eq_ignore_ascii_case("client")
}
struct UnknownAcc {
provider: String,
paths: std::collections::BTreeSet<String>,
files: std::collections::BTreeSet<String>,
}
struct MissingAcc {
path: String,
files: std::collections::BTreeSet<String>,
}
pub async fn check_tag_references(
zip: &[u8],
excluded: &[String],
gateway: &dyn ProjectGateway,
) -> TagCheck {
check_tag_references_with(zip, excluded, gateway, &TagReads::default()).await
}
#[derive(Debug, Default)]
pub struct TagReads {
pub names: Option<Vec<String>>,
pub exports: std::collections::BTreeMap<String, Vec<u8>>,
}
pub async fn check_tag_references_with(
zip: &[u8],
excluded: &[String],
gateway: &dyn ProjectGateway,
reads: &TagReads,
) -> TagCheck {
let refs = match scan_artifact(zip) {
Ok(refs) => refs,
Err(err) => {
return TagCheck {
missing: None,
unknown_providers: None,
unavailable: vec![TagsUnavailable {
provider: None,
reason: unavailable_reason("the project files could not be scanned", &err),
}],
};
}
};
let mut check = check_refs(&refs, excluded, gateway, reads).await;
if let Some(reason) = refs.cut_reason() {
check.unavailable.insert(
0,
TagsUnavailable {
provider: None,
reason,
},
);
check.missing = check.missing.filter(|list| !list.is_empty());
check.unknown_providers = check.unknown_providers.filter(|list| !list.is_empty());
}
check
}
async fn check_refs(
refs: &TagRefs,
excluded: &[String],
gateway: &dyn ProjectGateway,
reads: &TagReads,
) -> TagCheck {
let mut by_provider: std::collections::BTreeMap<String, Vec<&TagRef>> =
std::collections::BTreeMap::new();
for tag in refs.iter() {
let fold = tag.provider.to_ascii_lowercase();
if exclude::is_excluded(&fold, excluded) {
continue;
}
by_provider.entry(fold).or_default().push(tag);
}
if by_provider.is_empty() {
return TagCheck {
missing: Some(Vec::new()),
unknown_providers: Some(Vec::new()),
unavailable: Vec::new(),
};
}
let listed_names = match &reads.names {
Some(names) => Ok(names.clone()),
None => gateway.tag_provider_names().await,
};
let names = match listed_names {
Ok(names) => names,
Err(err) => {
return TagCheck {
missing: None,
unknown_providers: None,
unavailable: vec![TagsUnavailable {
provider: None,
reason: unavailable_reason(
"the gateway's tag providers could not be listed",
&err,
),
}],
};
}
};
let mut listed: std::collections::BTreeMap<String, String> = std::collections::BTreeMap::new();
for name in names {
let fold = name.to_ascii_lowercase();
match listed.get(&fold) {
Some(kept) if kept.as_bytes() <= name.as_bytes() => {}
_ => {
listed.insert(fold, name);
}
}
}
let mut found: std::collections::BTreeMap<(String, String), MissingAcc> =
std::collections::BTreeMap::new();
let mut unknown: std::collections::BTreeMap<String, UnknownAcc> =
std::collections::BTreeMap::new();
let mut unavailable = Vec::new();
let (mut needed, mut checked) = (0usize, 0usize);
for (fold, tags) in &by_provider {
let Some(provider) = listed.get(fold) else {
if provider_shaped(fold) {
let acc = unknown.entry(fold.clone()).or_insert_with(|| UnknownAcc {
provider: tags[0].provider.clone(),
paths: std::collections::BTreeSet::new(),
files: std::collections::BTreeSet::new(),
});
for tag in tags {
if tag.provider < acc.provider {
acc.provider = tag.provider.clone();
}
let path = strip_property(&tag.path).unwrap_or(&tag.path);
acc.paths.insert(path.to_lowercase());
acc.files.extend(tag.files.iter().cloned());
while acc.files.len() > FILES_PER_ENTRY {
acc.files.pop_last();
}
}
}
continue;
};
needed += 1;
let read = match reads.exports.get(provider) {
Some(bytes) => Ok(bytes.clone()),
None => gateway.tag_export(provider).await,
};
let bytes = match read {
Ok(bytes) => bytes,
Err(err) => {
unavailable.push(TagsUnavailable {
provider: Some(provider.clone()),
reason: unavailable_reason("the provider's tag export could not be read", &err),
});
continue;
}
};
let tree = match TagTree::parse(&bytes, provider) {
Ok(tree) => tree,
Err(err) => {
unavailable.push(TagsUnavailable {
provider: Some(provider.clone()),
reason: unavailable_reason("the provider's tag export is not a tag tree", &err),
});
continue;
}
};
drop(bytes);
let mut unresolved: std::collections::BTreeSet<String> = std::collections::BTreeSet::new();
for tag in tags {
let full = tree.lookup(&tag.path);
let answer = match (full, strip_property(&tag.path)) {
(Lookup::Found, _) => Lookup::Found,
(other, None) => other,
(full, Some(stripped)) => match (full, tree.lookup(stripped)) {
(_, Lookup::Found) => Lookup::Found,
(Lookup::Missing, Lookup::Missing) => Lookup::Missing,
_ => Lookup::Unknown,
},
};
let reported = match answer {
Lookup::Found => {
checked += 1;
continue;
}
Lookup::Unknown => {
unresolved.insert(format!("[{}]{}", tag.provider, tag.path));
continue;
}
Lookup::Missing => {
checked += 1;
strip_property(&tag.path).unwrap_or(&tag.path)
}
};
let display = format!("[{}]{}", tag.provider, reported);
let acc = found
.entry((fold.clone(), reported.to_lowercase()))
.or_insert_with(|| MissingAcc {
path: display.clone(),
files: std::collections::BTreeSet::new(),
});
if display < acc.path {
acc.path = display;
}
acc.files.extend(tag.files.iter().cloned());
while acc.files.len() > FILES_PER_ENTRY {
acc.files.pop_last();
}
}
if !unresolved.is_empty() {
unavailable.push(TagsUnavailable {
provider: Some(provider.clone()),
reason: unresolved_reason(&unresolved),
});
}
}
let missing = if needed > 0 && checked == 0 {
None
} else {
let mut list: Vec<MissingTag> = found
.into_values()
.map(|acc| MissingTag {
path: acc.path,
files: acc.files.into_iter().collect(),
})
.collect();
list.sort_by(|a, b| a.path.as_bytes().cmp(b.path.as_bytes()));
Some(list)
};
let mut unknown_providers: Vec<UnknownProvider> = unknown
.into_values()
.map(|acc| UnknownProvider {
provider: acc.provider,
path_count: acc.paths.len(),
files: acc.files.into_iter().collect(),
})
.collect();
unknown_providers.sort_by(|a, b| a.provider.as_bytes().cmp(b.provider.as_bytes()));
TagCheck {
missing,
unknown_providers: Some(unknown_providers),
unavailable,
}
}
const WARNING_PROVIDERS_SHOWN: usize = 5;
fn tag_warnings(project: &str, gateway: &str, check: &TagCheck) -> Vec<String> {
let mut warnings = Vec::new();
let missing = check.missing.as_ref().map_or(0, Vec::len);
if missing > 0 {
warnings.push(format!(
"project {project}: {missing} tag path(s) the project uses are not on gateway {gateway}; \
ign deploy projects does not deploy tags, and paths built at runtime aren't checked"
));
}
if let Some(unknown) = check.unknown_providers.as_ref().filter(|u| !u.is_empty()) {
let shown: Vec<&str> = unknown
.iter()
.take(WARNING_PROVIDERS_SHOWN)
.map(|u| u.provider.as_str())
.collect();
let rest = unknown.len().saturating_sub(WARNING_PROVIDERS_SHOWN);
warnings.push(format!(
"project {project}: tag paths the project uses name {} provider(s) that gateway {gateway} does not have: {}{}; \
ign deploy projects does not deploy tags, and paths built at runtime aren't checked",
unknown.len(),
shown.join(", "),
if rest > 0 {
format!(" and {rest} more")
} else {
String::new()
}
));
}
for entry in &check.unavailable {
warnings.push(format!(
"project {project}: tag paths were not checked for {}: {}",
entry.provider.as_deref().unwrap_or("every provider"),
entry.reason
));
}
warnings
}
fn parent_link(record: &ProjectRecord) -> ParentLink {
ParentLink {
name: record.name.clone(),
parent: record.parent.clone().filter(|p| !p.is_empty()),
inheritable: record.inheritable,
invalid_parent: record.invalid_parent,
}
}
fn live_digest(project: &str, zip: &[u8]) -> Result<CanonicalDigest, CoreError> {
canonical_digest(zip).map_err(|err| match err {
CoreError::DeployTreeInvalid(_) => CoreError::Internal(format!(
"the gateway's export of project {project} holds no members"
)),
other => other,
})
}
#[derive(Debug, Clone)]
pub enum ProjectsPrepared {
Ready {
prose: String,
plan: DeployProjectsResult,
tag_records: std::collections::BTreeMap<String, TagRecord>,
},
Unchanged(DeployProjectsResult),
}
impl ProjectsPrepared {
pub fn into_check_result(self) -> DeployProjectsResult {
let (ProjectsPrepared::Ready {
plan: mut result, ..
}
| ProjectsPrepared::Unchanged(mut result)) = self;
result.check = true;
result.outcome = ProjectOutcome::Checked;
result
}
}
fn plan_prose(plan: &DeployProjectsResult) -> String {
let action = match plan.action {
ProjectAction::Create => "create project",
ProjectAction::Import => "overwrite-import project",
ProjectAction::Record => "record the release only for project",
ProjectAction::Skip => "leave project",
};
let mut lines = vec![format!(
"deploy projects will {action} {} on gateway {} (env {}, release {})",
plan.project, plan.gateway, plan.env, plan.release
)];
if plan.action == ProjectAction::Record {
lines.push("the project itself is not changed".to_string());
} else {
lines.push(format!(
"members: {} added, {} changed, {} removed, {} same",
plan.members.added, plan.members.changed, plan.members.removed, plan.members.same
));
}
if plan.removed_count > 0 {
lines.push(format!(
"the import deletes {} member(s) the repo does not have:",
plan.removed_count
));
for path in plan.removed.iter().take(PROSE_REMOVED_SHOWN) {
lines.push(format!(" {path}"));
}
if plan.removed_count > PROSE_REMOVED_SHOWN {
lines.push(format!(
" ... and {} more",
plan.removed_count - PROSE_REMOVED_SHOWN
));
}
}
let missing = plan.missing_tags_count.unwrap_or(0);
if missing > 0 {
lines.push(format!(
"{missing} tag path(s) the project uses are not on gateway {} (this verb does not deploy tags):",
plan.gateway
));
for tag in plan.missing_tags.iter().flatten().take(PROSE_REMOVED_SHOWN) {
lines.push(format!(" {}", tag.path));
}
if missing > PROSE_REMOVED_SHOWN {
lines.push(format!(" ... and {} more", missing - PROSE_REMOVED_SHOWN));
}
}
let unknown = plan.missing_tags_unknown_providers_count.unwrap_or(0);
if unknown > 0 {
lines.push(format!(
"{unknown} provider(s) the project uses are not on gateway {}:",
plan.gateway
));
for entry in plan
.missing_tags_unknown_providers
.iter()
.flatten()
.take(PROSE_REMOVED_SHOWN)
{
lines.push(format!(" {}", entry.provider));
}
if unknown > PROSE_REMOVED_SHOWN {
lines.push(format!(" ... and {} more", unknown - PROSE_REMOVED_SHOWN));
}
}
for entry in plan.missing_tags_unavailable.iter().flatten() {
lines.push(format!(
"tags NOT CHECKED for {}: {}",
entry.provider.as_deref().unwrap_or("every provider"),
entry.reason
));
}
if let Some(tags) = &plan.tags {
lines.push(format!(
"owned tag providers ({}), deployed first, then the project, in one run:",
tags.providers.len()
));
for p in &tags.providers {
lines.push(format!(
" {}: {} added, {} changed, {} same, {} live-only (kept)",
p.provider, p.added, p.changed, p.same, p.drift
));
}
for p in &tags.providers {
let extra = [
(p.kept_values, "kept values (live value kept)"),
(p.removed, "removed"),
(p.already_absent, "already absent"),
(p.recreated, "recreated"),
(p.non_canonical, "non-canonical file finding(s)"),
];
let said: Vec<String> = extra
.iter()
.filter(|(n, _)| *n > 0)
.map(|(n, what)| format!("{n} {what}"))
.collect();
if !said.is_empty() {
lines.push(format!(" {}: {}", p.provider, said.join(", ")));
}
if p.first_change {
lines.push(format!(
" {}: first deploy of the provider by this tool (--allow-first-change)",
p.provider
));
} else if p.record_only {
lines.push(format!(
" {}: first deploy of the provider by this tool, every node equal",
p.provider
));
}
}
let list = |lines: &mut Vec<String>, verb: &str, paths: &[String], count: usize| {
for path in paths.iter().take(PROSE_REMOVED_SHOWN) {
lines.push(format!(" {verb} {path}"));
}
if count > PROSE_REMOVED_SHOWN {
lines.push(format!(
" ... and {} more {verb}",
count - PROSE_REMOVED_SHOWN
));
}
};
list(&mut lines, "add", &tags.added, tags.added_count);
let changed: Vec<String> = tags
.changed
.iter()
.map(|c| {
format!(
"{} ({}){}",
c.path,
c.properties.join(", "),
if c.restart_likely {
", restart likely"
} else {
""
}
)
})
.collect();
list(&mut lines, "change", &changed, tags.changed_count);
list(
&mut lines,
"remove (with --allow-destructive)",
&tags.removed,
tags.removed_count,
);
let groups: Vec<String> = tags
.groups
.iter()
.flat_map(|g| g.removed.iter().map(|n| full_tag_path(&g.provider, n)))
.collect();
list(
&mut lines,
"remove tag group (with --allow-destructive)",
&groups,
groups.len(),
);
list(
&mut lines,
"kept value",
&tags.kept_values,
tags.kept_values_count,
);
for warning in &tags.warnings {
lines.push(format!(" warning: {warning}"));
}
lines.push(format!("tags_sha256 {}", tags.tags_sha256));
lines.push(format!("tags plan_sha256 {}", tags.plan_sha256));
}
if plan.action == ProjectAction::Import {
lines.push("the live project is saved into a run directory before the import".to_string());
}
lines.push(format!("artifact_sha256 {}", plan.artifact_sha256));
lines.join("\n")
}
const PROSE_REMOVED_SHOWN: usize = 20;
pub async fn deploy_projects_prepare(
loaded: &LoadedProject,
input: &ProjectsInput,
surfaces: &ProjectsSurfaces<'_>,
) -> Result<ProjectsPrepared, CoreError> {
let project = &loaded.file.project;
let gateway = surfaces.gateway;
let gateway_name = &loaded.target.gateway;
let git = surfaces.host.git_entry(project).await?;
if git.is_present() {
return Err(CoreError::ProjectGitManaged {
project: project.clone(),
location: git_entry_location(project),
kind: git.as_str().to_string(),
});
}
let record = match surfaces
.host
.read_release_record(gateway_name, project)
.await?
{
None => None,
Some(bytes) => {
let location = release_record_location(
LedgerRoot::for_node(&loaded.target.nodes.master),
gateway_name,
project,
);
Some(ReleaseRecord::parse(
&bytes,
&location,
gateway_name,
project,
)?)
}
};
if let Some(record) = &record
&& loaded.release.order_against(&record.release()?) == ReleaseOrder::Older
&& !input.allow_downgrade
{
return Err(CoreError::DestructiveChangeRefused {
gate: "downgrade".to_string(),
keys: vec![project.clone()],
count: 1,
});
}
let tag_records = read_tag_records(loaded, surfaces.host).await?;
let mut warnings = Vec::new();
let journal = read_project_journal(loaded, surfaces.host).await?;
if let Some(journal) = &journal {
warnings.push(format!(
"project {project}: run {run} left its host journal ({state}) at {location}; an apply \
refuses until `{command}` undoes it",
run = journal.run_id,
state = journal.state.as_str(),
location = project_journal_location(
LedgerRoot::for_node(&loaded.target.nodes.master),
gateway_name,
project
),
command = rollback_run_command(&input.env, &journal.run_id),
));
}
let records = gateway.projects().await?;
if let Some(parent) = &loaded.meta.parent {
let all: Vec<ParentLink> = records.iter().map(parent_link).collect();
let (chain, bound_hit) = parent_chain(parent, &all);
check_parent(project, parent, &chain, bound_hit).map_err(|reason| {
CoreError::ProjectParentInvalid {
project: project.clone(),
parent: parent.clone(),
reason: reason.as_str().to_string(),
}
})?;
if chain
.first()
.is_some_and(|p| p.invalid_parent == Some(true))
{
warnings.push(format!(
"project {project}: the gateway reports that parent {parent} has an invalid parent of its own"
));
}
}
let repo_parent = loaded.meta.parent.as_deref().filter(|p| !p.is_empty());
if let Some(live_record) = records.iter().find(|r| r.name == *project) {
let live_parent = live_record.parent.as_deref().filter(|p| !p.is_empty());
if live_parent != repo_parent {
warnings.push(format!(
"project {project}: importing this project changes its parent from {} to {}",
live_parent.unwrap_or("none"),
repo_parent.unwrap_or("none")
));
}
}
let mut others: Vec<String> = records
.into_iter()
.map(|r| r.name)
.filter(|name| name != project)
.collect();
others.sort_by(|a, b| a.as_bytes().cmp(b.as_bytes()));
others.dedup();
let live = match gateway.project_find(project).await? {
None => None,
Some(_) => {
let zip = gateway.export_bytes(project).await?;
Some(live_digest(project, &zip)?)
}
};
let artifact = &loaded.artifact.digest;
let equal = live.as_ref().is_some_and(|live| {
!live.members.is_empty() && !artifact.members.is_empty() && live.sha256 == artifact.sha256
});
let action = match &live {
None => ProjectAction::Create,
Some(_) if equal => match &record {
Some(record) if record.release()?.version == loaded.release.version => {
ProjectAction::Skip
}
_ => ProjectAction::Record,
},
Some(_) => ProjectAction::Import,
};
if action == ProjectAction::Import && record.is_none() && input.allow_first_change.is_none() {
return Err(CoreError::DestructiveChangeRefused {
gate: "first_deploy".to_string(),
keys: vec![project.clone()],
count: 1,
});
}
let (tag_plan, tag_reads, provider_settings, tag_live) =
match plan_owned_tags(loaded, gateway, &tag_records).await? {
Some(owned) => (Some(owned.plan), owned.reads, owned.settings, owned.live),
None => (None, TagReads::default(), Vec::new(), ProviderReads::new()),
};
let tag_secret_warnings = match &tag_plan {
Some(plan) => tag_gates(loaded, input, plan)?,
None => Vec::new(),
};
let groups = match &tag_plan {
Some(plan) => plan_owned_groups(loaded, surfaces, plan, &tag_records, &tag_live).await?,
None => Vec::new(),
};
drop(tag_live);
let tag_check = check_tag_references_with(
&loaded.artifact.zip,
&loaded.tag_exclusions,
gateway,
&tag_reads,
)
.await;
drop(tag_reads);
let live_matches_last_deploy = match (&record, &live) {
(Some(record), Some(live)) => Some(record.artifact_sha256 == live.sha256),
_ => None,
};
if let (Some(record), Some(false)) = (&record, live_matches_last_deploy) {
warnings.push(format!(
"project {project}: the live project no longer matches the last release this tool deployed ({})",
record.release
));
}
warnings.extend(tag_warnings(project, gateway_name, &tag_check));
let (members, removed) = member_delta(artifact, live.as_ref());
let (removed, removed_count) = capped(removed);
let (excluded, excluded_count) = capped(loaded.artifact.excluded.clone());
let (unmanaged, unmanaged_count) = capped(others);
let (missing_tags, missing_tags_count) = match tag_check.missing {
Some(list) => {
let (list, count) = cap_list(list);
(Some(list), Some(count))
}
None => (None, None),
};
let (missing_tags_unknown_providers, missing_tags_unknown_providers_count) =
match tag_check.unknown_providers {
Some(list) => {
let (list, count) = cap_list(list);
(Some(list), Some(count))
}
None => (None, None),
};
let missing_tags_unavailable =
(!tag_check.unavailable.is_empty()).then_some(tag_check.unavailable);
let settings_change = provider_settings
.iter()
.any(|p| p.action != ProviderSettingsAction::Same);
let would_change = action.would_change()
|| tag_plan.as_ref().is_some_and(TagPlan::would_change)
|| settings_change
|| groups.iter().any(GroupPlan::would_change);
let delete_route = match (&tag_plan, &surfaces.tag_delete) {
(None, _) => None,
(Some(_), TagDeleteSurface::Configured(route)) => Some(RouteState::of(
&surfaces.tag_delete,
Some(route.handshake().await),
)),
(Some(_), surface) => Some(RouteState::of(surface, None)),
}
.map(|state| state.report(&owned_names(loaded)));
let tags = tag_plan.map(|plan| {
let mut report = TagsReport::from_plan(&plan, &loaded.non_canonical, LIST_CAP);
report.delete_route = delete_route;
report.warnings.clone_from(&loaded.ownership_warnings);
report.warnings.extend(tag_secret_warnings);
report.provider_settings = provider_settings;
report.groups = groups;
report.journal = journal.as_ref().map(|j| TagJournal {
run_id: j.run_id.clone(),
state: j.state.as_str().to_string(),
started_at: j.started_at.clone(),
});
Box::new(report)
});
let outcome = if !would_change {
ProjectOutcome::Skipped
} else if tags.is_some() {
ProjectOutcome::Applied
} else {
match action {
ProjectAction::Create => ProjectOutcome::Created,
ProjectAction::Skip => ProjectOutcome::Skipped,
ProjectAction::Import => ProjectOutcome::Imported,
ProjectAction::Record => ProjectOutcome::Recorded,
}
};
let plan = DeployProjectsResult {
env: input.env.clone(),
gateway: loaded.target.gateway.clone(),
project: project.clone(),
release: input.release.clone(),
check: false,
action,
would_change,
artifact_sha256: artifact.sha256.clone(),
live_sha256: live.map(|l| l.sha256),
members,
removed,
removed_count,
excluded,
excluded_count,
unmanaged,
unmanaged_count,
parent: loaded.meta.parent.clone(),
previous_release: record.as_ref().map(|r| r.release.clone()),
recorded_sha256: record.map(|r| r.artifact_sha256),
live_matches_last_deploy,
allow_downgrade: input.allow_downgrade,
allow_first_change: input.allow_first_change.clone(),
warnings,
run_id: None,
run_dir: None,
record_written: false,
restore_command: None,
children: Vec::new(),
missing_tags,
missing_tags_count,
missing_tags_unknown_providers,
missing_tags_unknown_providers_count,
missing_tags_unavailable,
missing_tags_limit: MISSING_TAGS_LIMIT,
tags,
outcome,
ok: true,
slug: None,
};
Ok(if !would_change {
ProjectsPrepared::Unchanged(plan)
} else {
ProjectsPrepared::Ready {
prose: plan_prose(&plan),
plan,
tag_records,
}
})
}
struct OwnedTags {
plan: TagPlan,
reads: TagReads,
live: ProviderReads,
settings: Vec<ProviderSettingsPlan>,
}
async fn plan_owned_tags(
loaded: &LoadedProject,
gateway: &dyn ProjectGateway,
records: &std::collections::BTreeMap<String, TagRecord>,
) -> Result<Option<OwnedTags>, CoreError> {
if loaded.owned_tags.is_empty() {
return Ok(None);
}
let names = gateway.tag_provider_names().await?;
let mut findings = Vec::new();
for (i, provider) in loaded.owned_tags.iter().enumerate() {
if names.contains(&provider.name)
|| !names.iter().any(|n| n.eq_ignore_ascii_case(&provider.name))
{
continue;
}
findings.push(crate::error::DeployFinding {
file: PROJECT_FILE_REL.to_string(),
line: None,
column: None,
message: format!(
"tag_providers[{i}]: gateway {} has no tag provider named {} (it lists one whose \
name differs only in letter case; the names must match exactly)",
loaded.target.gateway, provider.name
),
});
}
if !findings.is_empty() {
return Err(CoreError::deploy_findings(findings));
}
let mut settings = Vec::new();
for provider in sorted(&loaded.owned_tags) {
let listed = names.contains(&provider.name);
let record = if listed {
let record = gateway.tag_provider_record(&provider.name).await?;
if record.is_none() {
return Err(CoreError::Internal(format!(
"the gateway lists tag provider {} but has no record of it",
provider.name
)));
}
record
} else {
None
};
settings.push(plan_provider_settings(
&provider.name,
provider_settings_of(loaded, &provider.name)?,
record.as_ref(),
));
}
let mut reads = TagReads {
names: Some(names),
exports: std::collections::BTreeMap::new(),
};
let mut live = std::collections::BTreeMap::new();
for provider in &loaded.owned_tags {
if !reads
.names
.as_ref()
.is_some_and(|names| names.contains(&provider.name))
{
continue;
}
let bytes = gateway.tag_export(&provider.name).await?;
live.insert(
provider.name.clone(),
flatten_export(&bytes, &provider.name)?,
);
reads.exports.insert(provider.name.clone(), bytes);
}
let plan = plan_tags(&loaded.owned_tags, &live, records);
let empty = std::collections::BTreeMap::new();
let mut in_use = Vec::new();
for (provider_plan, provider) in plan.providers.iter().zip(sorted(&loaded.owned_tags)) {
let nodes = live.get(&provider.name).unwrap_or(&empty);
for (type_path, referrer) in removed_types_in_use(provider_plan, provider, nodes) {
in_use.push(crate::error::DeployFinding {
file: provider.nodes.get(&referrer).map_or_else(
|| full_tag_path(&provider.name, &referrer),
|node| node.file.clone(),
),
line: None,
column: None,
message: format!(
"the repo removes the type {} and {} still uses it (typeId), so deleting it \
would break a tag that stays; keep the type in the repo or remove its users \
too; nothing was written",
full_tag_path(&provider.name, &type_path),
full_tag_path(&provider.name, &referrer)
),
});
}
}
if !in_use.is_empty() {
return Err(CoreError::deploy_findings(in_use));
}
Ok(Some(OwnedTags {
plan,
reads,
settings,
live,
}))
}
fn need_tree<'a>(
surfaces: &ProjectsSurfaces<'a>,
what: &str,
) -> Result<&'a dyn TreeReader, CoreError> {
surfaces.tree.ok_or_else(|| {
CoreError::Internal(format!(
"{what} needs the master's host reader, and none was supplied"
))
})
}
const UNITS_SHOWN: usize = 10;
async fn refuse_mode_tag_units(
loaded: &LoadedProject,
reader: &dyn TreeReader,
) -> Result<(), CoreError> {
let mode = &loaded.target.mode;
let dirs: Vec<(String, String)> = crate::deploy::layout::PAIR_UNSYNCED_TAG_TYPES
.iter()
.map(|ty| ("ignition".to_string(), (*ty).to_string()))
.collect();
let listed = reader.list_tree(mode, &dirs).await?;
let mut found: Vec<String> = Vec::new();
for listing in &listed {
for provider in &loaded.file.tag_providers {
let root = format!("{}/{}/{provider}", listing.module, listing.ty);
let below = format!("{root}/");
found.extend(
listing
.entries
.iter()
.map(|e| e.path.as_str())
.filter(|p| *p == root || p.starts_with(&below))
.map(str::to_string),
);
}
}
if found.is_empty() {
return Ok(());
}
found.sort_by(|a, b| a.as_bytes().cmp(b.as_bytes()));
let count = found.len();
let shown: Vec<&str> = found.iter().take(UNITS_SHOWN).map(String::as_str).collect();
Err(CoreError::deploy_tree_invalid(
&loaded.target_file,
None,
None,
format!(
"nodes.master's {mode} collection holds {count} tag definition path(s) of an owned \
provider: {}{}. On a redundant pair the tag import would be written into the {mode} \
collection (C8), which stops the backup's configuration sync; move these units to \
core with a gateway-config deploy (ign deploy apply) first",
shown.join(", "),
if count > shown.len() {
format!(" and {} more", count - shown.len())
} else {
String::new()
}
),
))
}
struct GroupInputs<'a> {
provider: &'a NativeProvider,
summary: Option<std::collections::BTreeMap<String, GroupSummary>>,
sent: Vec<&'a NativeNode>,
removed: std::collections::BTreeSet<String>,
}
fn group_inputs<'a>(
loaded: &'a LoadedProject,
plan: &TagPlan,
) -> Result<Vec<GroupInputs<'a>>, CoreError> {
let mut out = Vec::new();
for (provider_plan, provider) in plan.providers.iter().zip(sorted(&loaded.owned_tags)) {
let summary = match &provider.groups {
Some(value) => Some(parse_group_value(
&format!("tags/{}/{TAG_GROUPS_FILE}", provider.name),
value,
)?),
None => None,
};
let sending: std::collections::BTreeSet<&str> = provider_plan
.added
.iter()
.map(String::as_str)
.chain(provider_plan.changed.iter().map(|c| c.path.as_str()))
.collect();
let sent = provider
.nodes
.values()
.filter(|n| sending.contains(n.path.as_str()))
.collect();
out.push(GroupInputs {
provider,
summary,
sent,
removed: provider_plan.removed.iter().cloned().collect(),
});
}
Ok(out)
}
fn group_plans(
inputs: &[GroupInputs<'_>],
observed: &crate::deploy::transport::ObservedFiles,
records: &std::collections::BTreeMap<String, TagRecord>,
live_nodes: &ProviderReads,
) -> Result<Vec<GroupPlan>, CoreError> {
let mut plans = Vec::new();
let mut in_use = Vec::new();
let empty = BTreeMap::new();
for input in inputs {
let Some(summary) = &input.summary else {
continue;
};
let name = &input.provider.name;
let live = LiveGroups::from_observed(name, observed);
let recorded = records
.get(name)
.map(|r| r.groups.clone())
.unwrap_or_default();
let referenced = referenced_groups(input.provider.nodes.values());
let plan = plan_groups(name, summary, &live, &recorded, &referenced)?;
if !plan.removed.is_empty() {
let users = live_group_users(live_nodes.get(name).unwrap_or(&empty), &input.removed);
for group in &plan.removed {
if let Some(user) = users.get(group) {
in_use.push(crate::error::DeployFinding {
file: format!("tags/{name}/{TAG_GROUPS_FILE}"),
line: None,
column: None,
message: format!(
"the repo drops tag group `{group}` of provider {name}, which this \
project recorded, and the live tag {} still names it, so deleting \
the group would leave that tag without its group; keep the group in \
the summary or change the tag first; nothing was written",
full_tag_path(name, user)
),
});
}
}
}
plans.push(plan);
}
if !in_use.is_empty() {
return Err(CoreError::deploy_findings(in_use));
}
Ok(plans)
}
fn providers_needing_groups(inputs: &[GroupInputs<'_>]) -> Vec<String> {
inputs
.iter()
.filter(|input| {
input.summary.is_some() || !referenced_groups(input.sent.iter().copied()).is_empty()
})
.map(|input| input.provider.name.clone())
.collect()
}
async fn read_groups(
reader: &dyn TreeReader,
mode: &str,
providers: &[String],
) -> Result<crate::deploy::transport::ObservedFiles, CoreError> {
let observed = reader.read_tag_groups(mode, providers).await?;
if let Some(first) = observed.unreadable.first() {
return Err(CoreError::TransportFailed {
operation: "read".to_string(),
node: "nodes.master".to_string(),
detail: format!(
"{} tag-group file(s) cannot be read, the first {}/{}",
observed.unreadable.len(),
first.collection,
first.path
),
});
}
Ok(observed)
}
async fn plan_owned_groups(
loaded: &LoadedProject,
surfaces: &ProjectsSurfaces<'_>,
plan: &TagPlan,
records: &std::collections::BTreeMap<String, TagRecord>,
live: &ProviderReads,
) -> Result<Vec<GroupPlan>, CoreError> {
if loaded.target.nodes.backup.is_some() {
refuse_mode_tag_units(
loaded,
need_tree(surfaces, "the pair's mode-collection check")?,
)
.await?;
}
let inputs = group_inputs(loaded, plan)?;
let providers = providers_needing_groups(&inputs);
if providers.is_empty() {
return Ok(Vec::new());
}
let reader = need_tree(surfaces, "the tag-group read")?;
let observed = read_groups(reader, &loaded.target.mode, &providers).await?;
let mut findings = Vec::new();
let empty = std::collections::BTreeMap::new();
for input in &inputs {
let live = LiveGroups::from_observed(&input.provider.name, &observed).names();
match check_group_refs(
&input.provider.name,
&input.sent,
input.summary.as_ref().unwrap_or(&empty),
&live,
) {
Ok(()) => {}
Err(CoreError::DeployTreeInvalid(details)) => findings.extend(details.findings),
Err(other) => return Err(other),
}
}
if !findings.is_empty() {
return Err(CoreError::deploy_findings(findings));
}
group_plans(&inputs, &observed, records, live)
}
fn provider_settings_of<'a>(
loaded: &'a LoadedProject,
provider: &str,
) -> Result<&'a ProviderSettings, CoreError> {
loaded
.file
.tag_providers
.iter()
.position(|name| name == provider)
.and_then(|i| loaded.provider_settings.get(i))
.ok_or_else(|| {
CoreError::Internal(format!(
"no settings were loaded for owned tag provider {provider}"
))
})
}
async fn read_tag_records(
loaded: &LoadedProject,
host: &dyn ProjectHost,
) -> Result<std::collections::BTreeMap<String, TagRecord>, CoreError> {
let gateway = &loaded.target.gateway;
let mut records = std::collections::BTreeMap::new();
for provider in &loaded.file.tag_providers {
let Some(bytes) = host.read_tag_record(gateway, provider).await? else {
continue;
};
let location = tag_record_location(
LedgerRoot::for_node(&loaded.target.nodes.master),
gateway,
provider,
);
let record = TagRecord::parse(&bytes, &location, gateway, provider)?;
if record.project != loaded.file.project {
return Err(CoreError::TagProviderClaimed {
provider: provider.clone(),
claimed_by: format!("project {} (host record)", record.project),
});
}
records.insert(provider.clone(), record);
}
Ok(records)
}
async fn read_project_journal(
loaded: &LoadedProject,
host: &dyn ProjectHost,
) -> Result<Option<ProjectJournal>, CoreError> {
if loaded.file.tag_providers.is_empty() {
return Ok(None);
}
let gateway = &loaded.target.gateway;
let project = &loaded.file.project;
let Some(bytes) = host.read_journal(gateway, project).await? else {
return Ok(None);
};
let location = project_journal_location(
LedgerRoot::for_node(&loaded.target.nodes.master),
gateway,
project,
);
ProjectJournal::parse(&bytes, &location, gateway, project).map(Some)
}
fn rollback_run_command(env: &str, run_id: &str) -> String {
format!(
"ign deploy projects --env {} --rollback-run {run_id}",
shell_word(env)
)
}
fn project_journal_present(loaded: &LoadedProject, env: &str, run_id: &str) -> CoreError {
CoreError::DeployJournalPresent {
run_id: run_id.to_string(),
env: env.to_string(),
location: project_journal_location(
LedgerRoot::for_node(&loaded.target.nodes.master),
&loaded.target.gateway,
&loaded.file.project,
),
recover_with: Some(rollback_run_command(env, run_id)),
}
}
fn tag_gates(
loaded: &LoadedProject,
input: &ProjectsInput,
plan: &TagPlan,
) -> Result<Vec<String>, CoreError> {
let named = |pick: fn(&crate::deploy::tag_plan::ProviderPlan) -> bool| -> Vec<String> {
plan.providers
.iter()
.filter(|p| pick(p))
.map(|p| format!("tag provider {}", p.provider))
.collect()
};
let emptied = named(|p| p.provider_emptied);
if !emptied.is_empty() {
return Err(CoreError::DestructiveChangeRefused {
gate: "provider_emptied".to_string(),
count: emptied.len(),
keys: emptied,
});
}
let first = named(|p| p.first_change);
if !first.is_empty() && input.allow_first_change.is_none() {
return Err(CoreError::DestructiveChangeRefused {
gate: "first_deploy".to_string(),
count: first.len(),
keys: first,
});
}
let mut refused = Vec::new();
let mut warnings = Vec::new();
for (provider_plan, provider) in plan.providers.iter().zip(sorted(&loaded.owned_tags)) {
let added: std::collections::BTreeSet<&str> =
provider_plan.added.iter().map(String::as_str).collect();
let changed: std::collections::BTreeSet<&str> = provider_plan
.changed
.iter()
.map(|c| c.path.as_str())
.collect();
for (path, node) in &provider.nodes {
let sends = if added.contains(path.as_str()) {
Sends::All
} else if changed.contains(path.as_str()) {
Sends::Config
} else {
Sends::Nothing
};
let scan = scan_tag_node(&node.file, path, &to_node(&node.json), sends);
refused.extend(scan.refused);
warnings.extend(scan.warnings);
}
}
if !refused.is_empty() {
return Err(CoreError::embedded_secret(refused, Vec::new()));
}
Ok(warnings)
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ProjectsEvent {
DesignersPoll {
open: usize,
},
DesignersProceed {
sessions: Vec<DesignerSeen>,
},
DesignersSkipped,
RunStarted {
run_id: String,
action: ProjectAction,
},
PreImageExported {
bytes: usize,
},
TagsPreImageExported {
provider: String,
bytes: usize,
},
TagImportSent {
provider: String,
path: String,
policy: &'static str,
nodes: usize,
carry: bool,
},
TagValuesCarried {
provider: String,
path: String,
nodes: usize,
},
TagProviderSettingsApplied {
provider: String,
action: &'static str,
},
TagGroupsWritten {
providers: Vec<String>,
written: usize,
deleted: usize,
},
TagImportAnswered {
provider: String,
clean: bool,
},
TagsVerified {
provider: String,
equal: bool,
},
ImportSent {
overwrite: bool,
},
ImportAnswered,
Verified {
equal: bool,
},
RecordWritten,
RecordFailed {
code: String,
},
TagRecordWritten {
provider: String,
},
TagRecordFailed {
provider: String,
code: String,
},
ChildrenSettling {
children: Vec<String>,
},
ChildChecked {
name: String,
status: ChildStatus,
},
RollbackStarted {
cause: RollbackCause,
},
ProjectRestored {
equal: bool,
},
TagGroupsRestored {
providers: Vec<String>,
},
TagProviderRestored {
provider: String,
action: &'static str,
},
RollbackVerified {
equal: bool,
},
TagDeleteSent {
provider: String,
paths: Vec<String>,
rollback: bool,
},
TagDeleteAnswered {
provider: String,
clean: bool,
deleted: usize,
},
HostPreImageWritten {
files: usize,
},
JournalWritten {
state: &'static str,
},
JournalCleared,
BackupWaitStarted {
budget_s: u64,
},
BackupResync {
ok: bool,
code: Option<String>,
},
BackupVerdict {
converged: bool,
reason: Option<&'static str>,
},
Finished {
outcome: ProjectOutcome,
},
}
impl ProjectsEvent {
pub fn name(&self) -> &'static str {
match self {
ProjectsEvent::DesignersPoll { .. } => "designers_poll",
ProjectsEvent::DesignersProceed { .. } => "designers_proceed",
ProjectsEvent::DesignersSkipped => "designers_skipped",
ProjectsEvent::RunStarted { .. } => "run_started",
ProjectsEvent::PreImageExported { .. } => "pre_image_exported",
ProjectsEvent::TagsPreImageExported { .. } => "tags_pre_image_exported",
ProjectsEvent::TagImportSent { .. } => "tag_import_sent",
ProjectsEvent::TagValuesCarried { .. } => "tag_values_carried",
ProjectsEvent::TagProviderSettingsApplied { .. } => "tag_provider_settings_applied",
ProjectsEvent::TagGroupsWritten { .. } => "tag_groups_written",
ProjectsEvent::TagImportAnswered { .. } => "tag_import_answered",
ProjectsEvent::TagsVerified { .. } => "tags_verified",
ProjectsEvent::ImportSent { .. } => "import_sent",
ProjectsEvent::ImportAnswered => "import_answered",
ProjectsEvent::Verified { .. } => "verified",
ProjectsEvent::RecordWritten => "record_written",
ProjectsEvent::RecordFailed { .. } => "record_failed",
ProjectsEvent::TagRecordWritten { .. } => "tag_record_written",
ProjectsEvent::TagRecordFailed { .. } => "tag_record_failed",
ProjectsEvent::ChildrenSettling { .. } => "children_settling",
ProjectsEvent::ChildChecked { .. } => "child_checked",
ProjectsEvent::RollbackStarted { .. } => "rollback_started",
ProjectsEvent::ProjectRestored { .. } => "project_restored",
ProjectsEvent::TagGroupsRestored { .. } => "tag_groups_restored",
ProjectsEvent::TagProviderRestored { .. } => "tag_provider_restored",
ProjectsEvent::RollbackVerified { .. } => "rollback_verified",
ProjectsEvent::TagDeleteSent { .. } => "tag_delete_sent",
ProjectsEvent::TagDeleteAnswered { .. } => "tag_delete_answered",
ProjectsEvent::HostPreImageWritten { .. } => "host_pre_image_written",
ProjectsEvent::JournalWritten { .. } => "journal_written",
ProjectsEvent::JournalCleared => "journal_cleared",
ProjectsEvent::BackupWaitStarted { .. } => "backup_wait_started",
ProjectsEvent::BackupResync { .. } => "backup_resync",
ProjectsEvent::BackupVerdict { .. } => "backup_verdict",
ProjectsEvent::Finished { .. } => "finished",
}
}
fn fields(&self) -> serde_json::Value {
match self {
ProjectsEvent::DesignersPoll { open } => serde_json::json!({ "open": open }),
ProjectsEvent::DesignersProceed { sessions } => {
serde_json::json!({ "sessions": sessions_json(sessions) })
}
ProjectsEvent::DesignersSkipped => serde_json::json!({ "reason": "record only" }),
ProjectsEvent::RunStarted { run_id, action } => {
serde_json::json!({ "run_id": run_id, "action": action })
}
ProjectsEvent::PreImageExported { bytes } => serde_json::json!({ "bytes": bytes }),
ProjectsEvent::TagsPreImageExported { provider, bytes } => {
serde_json::json!({ "provider": provider, "bytes": bytes })
}
ProjectsEvent::TagImportSent {
provider,
path,
policy,
nodes,
carry,
} => serde_json::json!({
"provider": provider, "path": path, "policy": policy, "nodes": nodes,
"carry": carry
}),
ProjectsEvent::TagValuesCarried {
provider,
path,
nodes,
} => serde_json::json!({ "provider": provider, "path": path, "nodes": nodes }),
ProjectsEvent::TagProviderSettingsApplied { provider, action } => {
serde_json::json!({ "provider": provider, "action": action })
}
ProjectsEvent::TagGroupsWritten {
providers,
written,
deleted,
} => serde_json::json!({
"providers": providers, "written": written, "deleted": deleted
}),
ProjectsEvent::TagImportAnswered { provider, clean } => {
serde_json::json!({ "provider": provider, "clean": clean })
}
ProjectsEvent::TagsVerified { provider, equal } => {
serde_json::json!({ "provider": provider, "equal": equal })
}
ProjectsEvent::ImportSent { overwrite } => {
serde_json::json!({ "overwrite": overwrite })
}
ProjectsEvent::Verified { equal } => serde_json::json!({ "equal": equal }),
ProjectsEvent::RecordFailed { code } => serde_json::json!({ "code": code }),
ProjectsEvent::TagRecordWritten { provider } => {
serde_json::json!({ "provider": provider })
}
ProjectsEvent::TagRecordFailed { provider, code } => {
serde_json::json!({ "provider": provider, "code": code })
}
ProjectsEvent::ChildrenSettling { children } => {
serde_json::json!({ "children": children })
}
ProjectsEvent::ChildChecked { name, status } => {
serde_json::json!({ "name": name, "status": status.as_str() })
}
ProjectsEvent::RollbackStarted { cause } => {
serde_json::json!({ "cause": cause.as_str() })
}
ProjectsEvent::ProjectRestored { equal }
| ProjectsEvent::RollbackVerified { equal } => {
serde_json::json!({ "equal": equal })
}
ProjectsEvent::TagGroupsRestored { providers } => {
serde_json::json!({ "providers": providers })
}
ProjectsEvent::TagProviderRestored { provider, action } => {
serde_json::json!({ "provider": provider, "action": action })
}
ProjectsEvent::TagDeleteSent {
provider,
paths,
rollback,
} => serde_json::json!({ "provider": provider, "paths": paths, "rollback": rollback }),
ProjectsEvent::TagDeleteAnswered {
provider,
clean,
deleted,
} => serde_json::json!({ "provider": provider, "clean": clean, "deleted": deleted }),
ProjectsEvent::HostPreImageWritten { files } => serde_json::json!({ "files": files }),
ProjectsEvent::JournalWritten { state } => serde_json::json!({ "state": state }),
ProjectsEvent::BackupWaitStarted { budget_s } => {
serde_json::json!({ "budget_s": budget_s })
}
ProjectsEvent::BackupResync { ok, code } => {
serde_json::json!({ "ok": ok, "code": code })
}
ProjectsEvent::BackupVerdict { converged, reason } => {
serde_json::json!({ "converged": converged, "reason": reason })
}
ProjectsEvent::Finished { outcome } => {
serde_json::json!({ "outcome": outcome.as_str() })
}
ProjectsEvent::ImportAnswered
| ProjectsEvent::RecordWritten
| ProjectsEvent::JournalCleared => serde_json::Value::Null,
}
}
}
pub trait ProjectsObserver: Send + Sync {
fn event(&self, event: &ProjectsEvent);
}
pub struct NoProjectsObserver;
impl ProjectsObserver for NoProjectsObserver {
fn event(&self, _event: &ProjectsEvent) {}
}
pub struct ProjectsRunOptions<'a> {
pub repo: PathBuf,
pub targets_root: PathBuf,
pub target_file: String,
pub master: NodeTarget,
#[doc(hidden)]
pub record_writer: Option<&'a dyn ReleaseRecordWriter>,
#[doc(hidden)]
pub tag_record_writer: Option<&'a dyn TagRecordWriter>,
#[doc(hidden)]
pub run_writer: Option<&'a dyn ProjectRunWriter>,
pub designers: DesignersPolicy,
pub children_settle: Duration,
pub children_poll: Duration,
pub tag_verify_settle: Duration,
pub tag_verify_poll: Duration,
pub backup: Option<&'a dyn BackupTags>,
pub backup_policy: BackupPolicy,
pub tag_delete_ready: Duration,
pub tag_delete_poll: Duration,
}
struct RunEvents<'a> {
log: EventLog,
observer: &'a dyn ProjectsObserver,
late_failure: bool,
}
impl RunEvents<'_> {
fn emit(&mut self, event: ProjectsEvent) -> Result<(), CoreError> {
self.observer.event(&event);
self.log.append(event.name(), event.fields())
}
fn record(&mut self, event: &ProjectsEvent) -> Result<(), CoreError> {
self.log.append(event.name(), event.fields())
}
fn emit_late(&mut self, event: ProjectsEvent) {
if self.emit(event).is_err() {
self.late_failure = true;
}
}
}
fn sessions_json(sessions: &[DesignerSeen]) -> Vec<serde_json::Value> {
sessions
.iter()
.map(|(user, project, host)| {
serde_json::json!({ "user": user, "project": project, "host": host })
})
.collect()
}
struct DesignersFound {
mode: &'static str,
timeout_s: Option<u64>,
polls: Vec<usize>,
seen: Vec<DesignerSeen>,
proceeded: bool,
skipped: bool,
}
async fn designers_check(
plan: &DeployProjectsResult,
gateway: &dyn ProjectGateway,
policy: &DesignersPolicy,
observer: &dyn ProjectsObserver,
) -> Result<DesignersFound, CoreError> {
if matches!(plan.action, ProjectAction::Record | ProjectAction::Skip) {
return Ok(DesignersFound {
mode: "skipped",
timeout_s: None,
polls: Vec::new(),
seen: Vec::new(),
proceeded: false,
skipped: true,
});
}
let mut polls = Vec::new();
let (verdict, seen) =
designers_gate(|| gateway.designers(), policy, &SystemClock, &mut |open| {
polls.push(open);
observer.event(&ProjectsEvent::DesignersPoll { open });
Ok(())
})
.await?;
if verdict == DesignersVerdict::Open {
return Err(CoreError::DeployPreflightFailed {
failures: Box::new([PreflightFailure {
check: "designers".to_string(),
detail: designers_detail(&seen),
}]),
});
}
Ok(DesignersFound {
mode: policy.mode.as_str(),
timeout_s: Some(policy.timeout.as_secs()),
polls,
seen,
proceeded: verdict == DesignersVerdict::Proceeded,
skipped: false,
})
}
async fn wrong_target_check(
loaded: &LoadedProject,
gateway: &dyn ProjectGateway,
opts: &ProjectsRunOptions<'_>,
) -> Result<(), CoreError> {
let rest = gateway.gateway_name().await?;
let reader = for_node(
&opts.targets_root,
&opts.target_file,
"nodes.master",
&opts.master,
)?;
let problems = wrong_target_gate(
rest.as_deref(),
reader.as_ref(),
&loaded.target.mode,
opts.master.gateway_name.as_deref(),
)
.await?;
if problems.is_empty() {
return Ok(());
}
Err(CoreError::DeployPreflightFailed {
failures: Box::new([wrong_target_failure(&problems)]),
})
}
fn release_record(plan: &DeployProjectsResult, release: &str, run_id: &RunId) -> ReleaseRecord {
ReleaseRecord {
schema_version: 1,
gateway: plan.gateway.clone(),
project: plan.project.clone(),
release: release.to_string(),
artifact_sha256: plan.artifact_sha256.clone(),
run_id: run_id.as_str().to_string(),
deployed_at: rfc3339_now_utc(),
}
}
fn provider_not_allowed(providers: &[&str]) -> CoreError {
CoreError::WebdevRouteError {
code: "provider_not_allowed".to_string(),
message: format!(
"the tag-delete route does not allow tag provider(s) {}, which this run removes tags \
from; reinstall the route from this project's repo with `ign webdev deploy \
--with-tag-delete --tag-delete-repo <repo> --tag-delete-user-source <name> \
--tag-delete-role <role>`; nothing was written",
providers.join(", ")
),
endpoint: None,
}
}
fn stale_live(expected: String, actual: String) -> CoreError {
CoreError::PlanStale {
expected,
actual,
detail: "live project changed since the plan".to_string(),
}
}
pub async fn deploy_projects_run(
prepared: ProjectsPrepared,
loaded: &LoadedProject,
input: &ProjectsInput,
surfaces: &ProjectsSurfaces<'_>,
opts: &ProjectsRunOptions<'_>,
observer: &dyn ProjectsObserver,
) -> Result<DeployProjectsResult, CoreError> {
let planned = match &prepared {
ProjectsPrepared::Unchanged(result) => result,
ProjectsPrepared::Ready { plan, .. } => plan,
};
if let Some(journal) = planned.tags.as_ref().and_then(|t| t.journal.as_ref()) {
return Err(project_journal_present(loaded, &input.env, &journal.run_id));
}
let (mut plan, tag_records) = match prepared {
ProjectsPrepared::Unchanged(result) => return Ok(result),
ProjectsPrepared::Ready {
plan, tag_records, ..
} => (plan, tag_records),
};
let removals = plan
.tags
.as_ref()
.is_some_and(|tags| tags.removed_count > 0);
let group_removals: Vec<String> = plan
.tags
.as_ref()
.map(|tags| {
tags.groups
.iter()
.flat_map(|g| {
g.removed
.iter()
.map(|name| format!("tag group {}", full_tag_path(&g.provider, name)))
})
.collect()
})
.unwrap_or_default();
if let Some(tags) = &plan.tags
&& (removals || !group_removals.is_empty())
&& !input.allow_destructive
{
let mut keys = tags.removed.clone();
keys.extend(group_removals.iter().cloned());
return Err(CoreError::DestructiveChangeRefused {
gate: "tag_removal".to_string(),
keys,
count: tags.removed_count + group_removals.len(),
});
}
if removals && let TagDeleteSurface::NotConfigured { missing } = &surfaces.tag_delete {
return Err(CoreError::TagDeleteNotConfigured {
profile: loaded.target.nodes.master.profile.clone(),
missing: missing.clone(),
});
}
let built;
let writer: &dyn ReleaseRecordWriter = match opts.record_writer {
Some(writer) => writer,
None => {
built = release_writer_for_node(
&opts.targets_root,
&opts.target_file,
"nodes.master",
&opts.master,
)?;
built.as_ref()
}
};
let built_tags;
let tag_writer: Option<&dyn TagRecordWriter> = match (&plan.tags, opts.tag_record_writer) {
(None, _) => None,
(Some(_), Some(writer)) => Some(writer),
(Some(_), None) => {
built_tags = tag_record_writer_for_node(
&opts.targets_root,
&opts.target_file,
"nodes.master",
&opts.master,
)?;
Some(built_tags.as_ref())
}
};
let built_run;
let run_writer: Option<&dyn ProjectRunWriter> = match (&plan.tags, opts.run_writer) {
(None, _) => None,
(Some(_), Some(writer)) => Some(writer),
(Some(_), None) => {
built_run = project_run_writer_for_node(
&opts.targets_root,
&opts.target_file,
"nodes.master",
&opts.master,
)?;
Some(built_run.as_ref())
}
};
let built_host;
let group_writes = plan
.tags
.as_ref()
.is_some_and(|tags| tags.groups.iter().any(GroupPlan::would_change));
let apply_host: Option<&dyn ApplyHost> = match surfaces.apply_host {
Some(host) => Some(host),
None if group_writes => {
built_host = apply_host_for_node(
&opts.targets_root,
&opts.target_file,
"nodes.master",
&opts.master,
&loaded.target.gateway,
&loaded.target.mode,
)?;
Some(built_host.as_ref())
}
None => None,
};
let surfaces = &ProjectsSurfaces {
gateway: surfaces.gateway,
host: surfaces.host,
tree: surfaces.tree,
apply_host,
deploy: surfaces.deploy,
tag_delete: surfaces.tag_delete.clone(),
};
wrong_target_check(loaded, surfaces.gateway, opts).await?;
let designers = designers_check(&plan, surfaces.gateway, &opts.designers, observer).await?;
let mut answer = match (&plan.tags, &surfaces.tag_delete) {
(Some(_), TagDeleteSurface::Configured(route)) => {
Some(wait_ready(*route, opts.tag_delete_ready, opts.tag_delete_poll).await)
}
_ => None,
};
if removals {
if let Some(Err(_)) = &answer
&& let Some(Err(err)) = answer.take()
{
return Err(err);
}
if let (Some(Ok(handshake)), Some(tags)) = (&answer, &plan.tags) {
let refused: Vec<&str> = tags
.providers
.iter()
.filter(|p| p.removed > 0)
.map(|p| p.provider.as_str())
.filter(|name| !handshake.providers.iter().any(|p| p == name))
.collect();
if !refused.is_empty() {
return Err(provider_not_allowed(&refused));
}
}
}
let route = RouteState::of(&surfaces.tag_delete, answer);
if let Some(tags) = plan.tags.as_mut() {
tags.delete_route = Some(route.report(&owned_names(loaded)));
if let (Some(_), None) = (route.route, &route.handshake) {
tags.warnings.push(format!(
"the tag-delete route is not ready ({}); a rollback cannot remove tags this run \
creates",
route.problem.as_deref().unwrap_or("no handshake")
));
}
}
let (run_id, run_dir) = RunDir::create_fresh(&opts.repo, &plan.gateway)?;
let rel = run_dir_rel(&plan.gateway, &run_id);
let outcome = execute(
&plan,
&tag_records,
loaded,
input,
surfaces,
opts,
Writers {
release: writer,
tags: tag_writer,
run: run_writer,
},
&run_dir,
&run_id,
&rel,
&designers,
&route,
observer,
)
.await;
match outcome {
Ok(mut result) => {
if let Err(err) = run_dir.write_json("result.json", &result) {
result.warnings.push(format!(
"the run's result.json could not be written ({})",
err.code()
));
}
Ok(result)
}
Err(err) => {
let _ = run_dir.write_json(
"result.json",
&serde_json::json!({
"ok": false,
"slug": err.code(),
"run_id": run_id.as_str(),
}),
);
Err(err)
}
}
}
struct ChildSweep {
children: Vec<ChildReport>,
warnings: Vec<String>,
}
type ChildRead = Result<String, String>;
async fn read_children(gateway: &dyn ProjectGateway, names: &[String]) -> Vec<ChildRead> {
let mut reads = Vec::with_capacity(names.len());
for name in names {
reads.push(match gateway.export_bytes(name).await {
Ok(zip) => live_digest(name, &zip)
.map(|digest| digest.sha256)
.map_err(|err| err.code().to_string()),
Err(err) => Err(err.code().to_string()),
});
}
reads
}
async fn sweep_children(
result: &DeployProjectsResult,
loaded: &LoadedProject,
surfaces: &ProjectsSurfaces<'_>,
opts: &ProjectsRunOptions<'_>,
events: &mut RunEvents<'_>,
) -> ChildSweep {
let project = result.project.as_str();
let gateway = surfaces.gateway;
let mut sweep = ChildSweep {
children: Vec::new(),
warnings: Vec::new(),
};
let records = match gateway.projects().await {
Ok(records) => records,
Err(err) => {
sweep.warnings.push(format!(
"could not list the projects to check the children of {project} ({})",
err.code()
));
return sweep;
}
};
let mut names: Vec<String> = records
.into_iter()
.filter(|record| record.parent.as_deref() == Some(project))
.map(|record| record.name)
.collect();
names.sort_by(|a, b| a.as_bytes().cmp(b.as_bytes()));
names.dedup();
if names.is_empty() {
return sweep;
}
events.emit_late(ProjectsEvent::ChildrenSettling {
children: names.clone(),
});
let started = Instant::now();
let mut current = read_children(gateway, &names).await;
let mut unsettled = vec![true; names.len()];
loop {
let elapsed = started.elapsed();
if elapsed >= opts.children_settle {
break;
}
tokio::time::sleep(opts.children_poll.min(opts.children_settle - elapsed)).await;
let next = read_children(gateway, &names).await;
unsettled = current.iter().zip(&next).map(|(a, b)| a != b).collect();
current = next;
if !unsettled.iter().any(|changing| *changing) {
break;
}
}
let gateway_name = result.gateway.as_str();
let root = LedgerRoot::for_node(&loaded.target.nodes.master);
for ((name, read), changing) in names.iter().zip(current).zip(unsettled) {
let live = match read {
Ok(sha256) => Some(sha256),
Err(code) => {
sweep.warnings.push(format!(
"child project {name}: its export could not be read ({code})"
));
None
}
};
if changing {
sweep.warnings.push(format!(
"child project {name} did not settle within {} s",
opts.children_settle.as_secs()
));
}
let record = match surfaces.host.read_release_record(gateway_name, name).await {
Ok(None) => None,
Ok(Some(bytes)) => {
let location = release_record_location(root, gateway_name, name);
match ReleaseRecord::parse(&bytes, &location, gateway_name, name) {
Ok(record) => Some(record),
Err(err) => {
sweep.warnings.push(format!(
"child project {name}: its release record could not be read ({})",
err.code()
));
None
}
}
}
Err(err) => {
sweep.warnings.push(format!(
"child project {name}: its release record could not be read ({})",
err.code()
));
None
}
};
let status = match (&record, &live) {
(Some(record), Some(live)) if record.artifact_sha256 == *live => ChildStatus::Ok,
(Some(record), Some(_)) => {
sweep.warnings.push(format!(
"child project {name} no longer matches the release this tool last deployed to it ({})",
record.release
));
ChildStatus::Mismatch
}
_ => ChildStatus::Unverified,
};
events.emit_late(ProjectsEvent::ChildChecked {
name: name.clone(),
status,
});
sweep.children.push(ChildReport {
name: name.clone(),
status,
release: record.as_ref().map(|r| r.release.clone()),
recorded_sha256: record.map(|r| r.artifact_sha256),
live_sha256: live,
});
}
sweep
}
#[allow(clippy::too_many_arguments)]
async fn execute(
plan: &DeployProjectsResult,
tag_records: &std::collections::BTreeMap<String, TagRecord>,
loaded: &LoadedProject,
input: &ProjectsInput,
surfaces: &ProjectsSurfaces<'_>,
opts: &ProjectsRunOptions<'_>,
writers: Writers<'_>,
run_dir: &RunDir,
run_id: &RunId,
rel: &str,
designers: &DesignersFound,
route: &RouteState<'_>,
observer: &dyn ProjectsObserver,
) -> Result<DeployProjectsResult, CoreError> {
let project = plan.project.as_str();
let gateway = surfaces.gateway;
let artifact = &loaded.artifact.digest;
let writer = writers.release;
run_dir.write_json(
"run.json",
&serde_json::json!({
"run_id": run_id.as_str(),
"started_at": rfc3339_now_utc(),
"env": plan.env,
"gateway": plan.gateway,
"project": plan.project,
"release": plan.release,
"action": plan.action,
"artifact_sha256": plan.artifact_sha256,
"live_sha256": plan.live_sha256,
"allow_downgrade": input.allow_downgrade,
"allow_first_change": input.allow_first_change,
"tag_providers": loaded.file.tag_providers,
"designers": designers.mode,
"designers_timeout_s": designers.timeout_s,
"designers_seen": sessions_json(&designers.seen),
"run_dir_git": run_dir.git_status().as_record(),
}),
)?;
run_dir.write_json("plan.json", plan)?;
let mut events = RunEvents {
log: EventLog::open(run_dir)?,
observer,
late_failure: false,
};
events.emit(ProjectsEvent::RunStarted {
run_id: run_id.as_str().to_string(),
action: plan.action,
})?;
for open in &designers.polls {
events.record(&ProjectsEvent::DesignersPoll { open: *open })?;
}
if designers.proceeded {
events.emit(ProjectsEvent::DesignersProceed {
sessions: designers.seen.clone(),
})?;
}
if designers.skipped {
events.emit(ProjectsEvent::DesignersSkipped)?;
}
let mut result = plan.clone();
result.run_id = Some(run_id.as_str().to_string());
result.run_dir = Some(rel.to_string());
let record = release_record(plan, &loaded.release.tag, run_id);
let tag_owning = plan.tags.is_some();
let backup_since_ms = wall_clock_ms();
let owned_names = owned_names(loaded);
if plan.action == ProjectAction::Record {
let zip = gateway.export_bytes(project).await?;
let live = live_digest(project, &zip)?;
if live.sha256 != artifact.sha256 {
return Err(stale_live(artifact.sha256.clone(), live.sha256));
}
events.emit(ProjectsEvent::Verified { equal: true })?;
let step = match deploy_owned_tags(
plan,
tag_records,
loaded,
input,
surfaces,
opts,
run_dir,
run_id,
HostWrites {
writer: writers.run,
project_zip: None,
},
route,
&mut events,
)
.await
{
Ok(step) => step,
Err(TagStepError::Unwritten(err)) => return Err(err),
Err(TagStepError::Written(failure)) => {
return roll_back(
*failure,
None,
result,
loaded,
input,
surfaces,
opts,
run_id,
rel,
UndoWriters {
run: writers.run,
records: None,
release: None,
},
route,
&mut events,
)
.await;
}
};
note_tag_step(&mut result, &step);
let changed = step.writes.wrote();
let mut record_written = true;
if let Err(err) = writer
.write_release_record(&plan.gateway, project, &record.to_bytes())
.await
{
events.emit_late(ProjectsEvent::RecordFailed {
code: err.code().to_string(),
});
if !changed {
return Err(unwritten(
err,
loaded,
writers.run,
step.writes.host.as_ref(),
&mut events,
)
.await);
}
record_written = false;
result.warnings.push(format!(
"the tags are verified but the release record could not be written ({})",
err.code()
));
}
if let Err(err) = write_tag_records(
&mut result,
loaded,
run_id,
writers.tags,
&mut events,
step.batches_sent + step.other_writes > 0,
)
.await
{
if changed {
return Err(err);
}
return Err(unwritten(
err,
loaded,
writers.run,
step.writes.host.as_ref(),
&mut events,
)
.await);
}
clear_host_journal(
&mut result,
loaded,
writers.run,
step.writes.host.as_ref(),
&mut events,
)
.await;
let backup = backup_step(
&mut result,
&owned_names,
surfaces,
opts,
backup_since_ms,
&mut events,
)
.await;
finish_recorded(
&mut result,
&mut events,
backup == Some(false),
record_written,
);
return Ok(result);
}
let overwrite = plan.action == ProjectAction::Import;
let mut pre_zip: Option<Vec<u8>> = None;
if overwrite {
let pre = gateway.export_bytes(project).await?;
let live = live_digest(project, &pre)?;
let mut file = run_dir.create_private_file_in("export", &format!("{project}.zip"))?;
file.write_all(&pre)
.and_then(|()| file.sync_all())
.map_err(|e| CoreError::DeployRunInvalid {
run_id: run_id.as_str().to_string(),
problem: format!("could not write the pre-image export ({:?})", e.kind()),
})?;
events.emit(ProjectsEvent::PreImageExported { bytes: pre.len() })?;
if plan.live_sha256.as_deref() != Some(live.sha256.as_str()) {
return Err(stale_live(
plan.live_sha256
.clone()
.unwrap_or_else(|| "none".to_string()),
live.sha256,
));
}
pre_zip = Some(pre);
}
let step = match deploy_owned_tags(
plan,
tag_records,
loaded,
input,
surfaces,
opts,
run_dir,
run_id,
HostWrites {
writer: writers.run,
project_zip: pre_zip.as_deref(),
},
route,
&mut events,
)
.await
{
Ok(step) => step,
Err(TagStepError::Unwritten(err)) => return Err(err),
Err(TagStepError::Written(failure)) => {
return roll_back(
*failure,
None,
result,
loaded,
input,
surfaces,
opts,
run_id,
rel,
UndoWriters {
run: writers.run,
records: None,
release: None,
},
route,
&mut events,
)
.await;
}
};
note_tag_step(&mut result, &step);
let tags_wrote = step.writes.wrote();
if plan.action == ProjectAction::Skip {
if let Err(err) = write_tag_records(
&mut result,
loaded,
run_id,
writers.tags,
&mut events,
step.batches_sent + step.other_writes > 0,
)
.await
{
if tags_wrote {
return Err(err);
}
return Err(unwritten(
err,
loaded,
writers.run,
step.writes.host.as_ref(),
&mut events,
)
.await);
}
clear_host_journal(
&mut result,
loaded,
writers.run,
step.writes.host.as_ref(),
&mut events,
)
.await;
result.outcome = ProjectOutcome::Applied;
if backup_step(
&mut result,
&owned_names,
surfaces,
opts,
backup_since_ms,
&mut events,
)
.await
== Some(false)
{
mark_backup_unsynced(&mut result);
}
events.emit_late(ProjectsEvent::Finished {
outcome: result.outcome,
});
note_late_log_failure(&mut result, &events);
return Ok(result);
}
if tag_owning && !designers.skipped {
match designers_check(plan, gateway, &opts.designers, observer).await {
Ok(found) => {
if found.proceeded {
events.emit_late(ProjectsEvent::DesignersProceed {
sessions: found.seen.clone(),
});
}
}
Err(err) if tags_wrote => {
let failure = TagFailure {
cause: RollbackCause::ProjectImportFailed,
what: format!(
"the Designer check right before the project import did not pass ({err})"
),
writes: step.writes,
};
return roll_back(
failure,
Some(ProjectUndo { pre: pre_zip }),
result,
loaded,
input,
surfaces,
opts,
run_id,
rel,
UndoWriters {
run: writers.run,
records: None,
release: None,
},
route,
&mut events,
)
.await;
}
Err(err) => {
return Err(unwritten(
err,
loaded,
writers.run,
step.writes.host.as_ref(),
&mut events,
)
.await);
}
}
}
if tag_owning {
events.emit_late(ProjectsEvent::ImportSent { overwrite });
} else {
events.emit(ProjectsEvent::ImportSent { overwrite })?;
}
let answer = match gateway
.project_import(project, loaded.artifact.zip.clone(), overwrite)
.await
{
Ok(answer) => Some(answer),
Err(err) if import_outcome_unknown(&err) => {
result.warnings.push(format!(
"the import request failed after it may have reached the gateway ({}); the project was read back to find out what it holds",
err.code()
));
None
}
Err(err) if tags_wrote => {
let failure = TagFailure {
cause: RollbackCause::ProjectImportFailed,
what: format!("the project import was refused ({})", err.code()),
writes: step.writes,
};
return roll_back(
failure,
Some(ProjectUndo { pre: pre_zip }),
result,
loaded,
input,
surfaces,
opts,
run_id,
rel,
UndoWriters {
run: writers.run,
records: None,
release: None,
},
route,
&mut events,
)
.await;
}
Err(err) => {
return Err(unwritten(
err,
loaded,
writers.run,
step.writes.host.as_ref(),
&mut events,
)
.await);
}
};
if let Some(answer) = &answer {
events.emit_late(ProjectsEvent::ImportAnswered);
if answer
.response
.get("changes")
.and_then(serde_json::Value::as_array)
.is_some_and(Vec::is_empty)
{
result
.warnings
.push("the gateway reported no changes for this import".to_string());
}
}
let equal = match retry_transient(|| gateway.export_bytes(project)).await {
Ok(zip) => match live_digest(project, &zip) {
Ok(live) => Some(live.sha256 == artifact.sha256),
Err(err) => {
result.warnings.push(format!(
"could not verify the import: the project read back is unusable ({})",
err.code()
));
None
}
},
Err(err) => {
result.warnings.push(format!(
"could not verify the import: the project could not be read back ({})",
err.code()
));
None
}
};
events.emit_late(ProjectsEvent::Verified {
equal: equal == Some(true),
});
if equal == Some(true) {
match writer
.write_release_record(&plan.gateway, project, &record.to_bytes())
.await
{
Ok(()) => {
events.emit_late(ProjectsEvent::RecordWritten);
result.record_written = true;
}
Err(err) => {
events.emit_late(ProjectsEvent::RecordFailed {
code: err.code().to_string(),
});
result.warnings.push(format!(
"the import is verified but the release record could not be written ({})",
err.code()
));
}
}
let _ =
write_tag_records(&mut result, loaded, run_id, writers.tags, &mut events, true).await;
clear_host_journal(
&mut result,
loaded,
writers.run,
step.writes.host.as_ref(),
&mut events,
)
.await;
if let Some(parent) = &loaded.meta.parent {
match gateway.projects().await {
Ok(records) => {
if records
.iter()
.find(|r| r.name == project)
.is_some_and(|r| r.invalid_parent == Some(true))
{
result.warnings.push(format!(
"project {project}: after the import the gateway reports that its parent {parent} is invalid"
));
}
}
Err(err) => result.warnings.push(format!(
"project {project}: the parent could not be read back after the import ({})",
err.code()
)),
}
}
result.outcome = if tag_owning {
ProjectOutcome::Applied
} else if overwrite {
ProjectOutcome::Imported
} else {
ProjectOutcome::Created
};
} else if tags_wrote {
let failure = TagFailure {
cause: if answer.is_none() {
RollbackCause::ProjectImportFailed
} else {
RollbackCause::ProjectUnverified
},
what: if equal == Some(false) {
"the project read back does not equal the artifact".to_string()
} else {
"the project could not be read back or compared after its import".to_string()
},
writes: step.writes,
};
return roll_back(
failure,
Some(ProjectUndo { pre: pre_zip }),
result,
loaded,
input,
surfaces,
opts,
run_id,
rel,
UndoWriters {
run: writers.run,
records: None,
release: None,
},
route,
&mut events,
)
.await;
} else {
result.outcome = ProjectOutcome::ImportedUnverified;
result.ok = false;
result.slug = Some(PROJECT_UNVERIFIED_SLUG.to_string());
if overwrite {
result.restore_command = Some(restore_command(project, rel));
}
if step.writes.host.is_some() {
result.warnings.push(format!(
"the host journal {} stays; `{}` restores the project from the host copy",
host_journal_location(loaded),
rollback_run_command(&input.env, run_id.as_str())
));
}
}
let swept = match result.outcome {
ProjectOutcome::Imported | ProjectOutcome::ImportedUnverified => true,
ProjectOutcome::Applied => overwrite,
_ => false,
};
if swept {
let sweep = sweep_children(&result, loaded, surfaces, opts, &mut events).await;
result.children = sweep.children;
result.warnings.extend(sweep.warnings);
}
if result.outcome == ProjectOutcome::Applied
&& backup_step(
&mut result,
&owned_names,
surfaces,
opts,
backup_since_ms,
&mut events,
)
.await
== Some(false)
{
mark_backup_unsynced(&mut result);
}
events.emit_late(ProjectsEvent::Finished {
outcome: result.outcome,
});
note_late_log_failure(&mut result, &events);
Ok(result)
}
const TAG_FAILURE_PATHS_SHOWN: usize = 20;
fn full_tag_path(provider: &str, path: &str) -> String {
format!("[{provider}]{path}")
}
struct TagStep {
carried: Vec<String>,
verify: Option<TagVerify>,
batches_sent: usize,
other_writes: usize,
writes: TagWrites,
deleted: Vec<String>,
absent: Vec<String>,
}
#[derive(Clone, Copy)]
struct Writers<'a> {
release: &'a dyn ReleaseRecordWriter,
tags: Option<&'a dyn TagRecordWriter>,
run: Option<&'a dyn ProjectRunWriter>,
}
type ProviderReads = BTreeMap<String, BTreeMap<String, Value>>;
async fn read_providers(
gateway: &dyn ProjectGateway,
names: &[&str],
) -> Result<ProviderReads, CoreError> {
let mut out = BTreeMap::new();
for name in names {
out.insert((*name).to_string(), read_flat_once(gateway, name).await?);
}
Ok(out)
}
async fn read_flat(
gateway: &dyn ProjectGateway,
name: &str,
) -> Result<BTreeMap<String, Value>, CoreError> {
let bytes = retry_transient(|| gateway.tag_export(name)).await?;
flatten_export(&bytes, name)
}
async fn read_flat_once(
gateway: &dyn ProjectGateway,
name: &str,
) -> Result<BTreeMap<String, Value>, CoreError> {
let bytes = gateway.tag_export(name).await?;
flatten_export(&bytes, name)
}
fn reads_agree(a: &ProviderReads, b: &ProviderReads) -> bool {
a.len() == b.len()
&& a.iter().all(|(provider, nodes)| {
b.get(provider).is_some_and(|other| {
nodes.len() == other.len()
&& nodes.iter().all(|(path, node)| {
other
.get(path)
.is_some_and(|o| diff_properties(node, o, path, EXPORTS).is_empty())
})
})
})
}
const EXPORTS: CanonOpts = CanonOpts {
strip_values: true,
created: false,
exports: true,
};
const RECREATED: CanonOpts = CanonOpts {
strip_values: true,
created: true,
exports: false,
};
fn recreated_at(path: &str, tops: &[String]) -> bool {
tops.iter().any(|top| {
path == top
|| (path.len() > top.len() + 1
&& path.starts_with(top.as_str())
&& path.as_bytes()[top.len()] == b'/')
})
}
async fn settled_reads(
gateway: &dyn ProjectGateway,
names: &[&str],
settle: Duration,
poll: Duration,
) -> Result<(ProviderReads, usize, bool, f64), CoreError> {
let started = Instant::now();
let mut reads = 1;
let mut last = retry_transient(|| read_providers(gateway, names)).await?;
let mut settled = false;
loop {
let elapsed = started.elapsed();
if elapsed >= settle {
break;
}
tokio::time::sleep(poll.min(settle - elapsed)).await;
reads += 1;
let next = match read_providers(gateway, names).await {
Ok(next) => next,
Err(err) if read_transient(&err) => continue,
Err(err) => return Err(err),
};
let agree = reads_agree(&last, &next);
last = next;
if agree {
settled = true;
break;
}
}
let elapsed_s = (started.elapsed().as_millis() as f64) / 1000.0;
Ok((last, reads, settled, elapsed_s))
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum RollbackCause {
TagStepFailed,
TagVerifyFailed,
ProjectImportFailed,
ProjectUnverified,
RollbackRun,
}
impl RollbackCause {
pub fn as_str(self) -> &'static str {
match self {
RollbackCause::TagStepFailed => "tag_step_failed",
RollbackCause::TagVerifyFailed => "tag_verify_failed",
RollbackCause::ProjectImportFailed => "project_import_failed",
RollbackCause::ProjectUnverified => "project_unverified",
RollbackCause::RollbackRun => "rollback_run",
}
}
}
enum GroupsDone {
Committed {
record: Box<RunRecord>,
providers: Vec<String>,
},
Restored,
Unrestored(String),
}
#[derive(Default)]
struct TagWrites {
plan: Option<TagPlan>,
pre: ProviderReads,
created: BTreeMap<String, CreatedProvider>,
modified: Vec<(String, TagProviderRecord)>,
applied: BTreeMap<String, AppliedSteps>,
groups: Option<GroupsDone>,
sent: Vec<String>,
carried: Vec<String>,
verify: Option<TagVerify>,
host: Option<HostCopy>,
recreate: BTreeMap<String, BTreeSet<String>>,
records_back: Option<Vec<(String, Option<Vec<u8>>)>>,
release_back: Option<Option<Vec<u8>>>,
delete_absent: Vec<String>,
}
#[derive(Clone, Copy)]
struct UndoWriters<'a> {
run: Option<&'a dyn ProjectRunWriter>,
records: Option<&'a dyn TagRecordWriter>,
release: Option<&'a dyn ReleaseRecordWriter>,
}
struct HostCopy {
run: PreImageRun,
journal: ProjectJournal,
}
#[derive(Clone, Copy)]
struct HostWrites<'a> {
writer: Option<&'a dyn ProjectRunWriter>,
project_zip: Option<&'a [u8]>,
}
struct PreImageInputs<'a> {
project_zip: Option<&'a [u8]>,
release: Option<&'a [u8]>,
exports: &'a BTreeMap<String, Vec<u8>>,
settings: &'a [(ProviderSettingsPlan, Option<TagProviderRecord>)],
records: &'a BTreeMap<String, TagRecord>,
plan: &'a TagPlan,
groups: bool,
}
fn action_word(action: ProjectAction) -> &'static str {
match action {
ProjectAction::Create => "create",
ProjectAction::Skip => "skip",
ProjectAction::Import => "import",
ProjectAction::Record => "record",
}
}
async fn write_host_copy(
writer: &dyn ProjectRunWriter,
plan: &DeployProjectsResult,
loaded: &LoadedProject,
run_id: &RunId,
pre: &PreImageInputs<'_>,
events: &mut RunEvents<'_>,
) -> Result<HostCopy, CoreError> {
let gateway = &loaded.target.gateway;
let project = &loaded.file.project;
let mut blobs: Vec<(String, std::borrow::Cow<'_, [u8]>)> = Vec::new();
if let Some(zip) = pre.project_zip {
blobs.push(("project.zip".to_string(), zip.into()));
}
if let Some(release) = pre.release {
blobs.push((PreImageFile::Release.file_name(), release.into()));
}
let owned = sorted(&loaded.owned_tags);
for provider in &owned {
let name = provider.name.as_str();
if let Some(bytes) = pre.exports.get(name) {
blobs.push((format!("tags-{name}.json"), bytes.as_slice().into()));
}
if let Some((_, Some(record))) = pre.settings.iter().find(|(p, _)| p.provider == name) {
let bytes = serde_json::to_vec_pretty(record).map_err(|e| {
CoreError::Internal(format!("tag provider {name}: its record: {e}"))
})?;
blobs.push((format!("provider-{name}.json"), bytes.into()));
}
if let Some(record) = pre.records.get(name) {
blobs.push((format!("record-{name}.json"), record.to_bytes().into()));
}
if pre.exports.contains_key(name)
&& let Some(provider_plan) = pre.plan.providers.iter().find(|p| p.provider == name)
&& !provider_plan.added.is_empty()
{
let sent: BTreeMap<&str, &Value> = provider_plan
.added
.iter()
.filter_map(|path| {
provider
.nodes
.get(path)
.map(|node| (path.as_str(), &node.json))
})
.collect();
let bytes = serde_json::to_vec(&sent).map_err(|e| {
CoreError::Internal(format!("tag provider {name}: the sent nodes: {e}"))
})?;
blobs.push((format!("sent-{name}.json"), bytes.into()));
}
}
let mut files = BTreeMap::new();
for (name, bytes) in &blobs {
writer
.write_pre_image(gateway, project, run_id.as_str(), name, bytes)
.await?;
files.insert(name.clone(), sha256_hex(bytes));
}
drop(blobs);
let providers = owned
.iter()
.map(|provider| {
let name = provider.name.as_str();
let provider_plan = pre.plan.providers.iter().find(|p| p.provider == name);
let mut added = provider_plan.map(|p| p.added.clone()).unwrap_or_default();
let mut changed: Vec<String> = provider_plan
.map(|p| p.changed.iter().map(|c| c.path.clone()).collect())
.unwrap_or_default();
for list in [&mut added, &mut changed] {
list.sort_by(|a, b| a.as_bytes().cmp(b.as_bytes()));
list.dedup();
}
let mut removed: Vec<String> =
provider_plan.map(|p| p.removed.clone()).unwrap_or_default();
removed.sort_by(|a, b| a.as_bytes().cmp(b.as_bytes()));
removed.dedup();
PreImageProvider {
name: name.to_string(),
settings: pre
.settings
.iter()
.find(|(p, _)| p.provider == name)
.map_or("same", |(p, _)| p.action.as_str())
.to_string(),
added,
changed,
created: None,
removed,
deleted: Vec::new(),
sent: Some(Vec::new()),
collided: Vec::new(),
}
})
.collect();
let run = PreImageRun {
schema_version: 1,
gateway: gateway.clone(),
project: project.clone(),
run_id: run_id.as_str().to_string(),
release: loaded.release.tag.clone(),
written_at: rfc3339_now_utc(),
project_action: action_word(plan.action).to_string(),
groups: pre.groups,
providers,
files,
};
writer
.write_pre_image(
gateway,
project,
run_id.as_str(),
"run.json",
&run.to_bytes(),
)
.await?;
events.emit(ProjectsEvent::HostPreImageWritten {
files: run.files.len() + 1,
})?;
let journal = ProjectJournal {
schema_version: 1,
gateway: gateway.clone(),
project: project.clone(),
run_id: run_id.as_str().to_string(),
state: JournalState::Applying,
started_at: rfc3339_now_utc(),
providers: owned.iter().map(|p| p.name.clone()).collect(),
};
writer
.write_journal(gateway, project, &journal.to_bytes())
.await?;
events.emit(ProjectsEvent::JournalWritten {
state: JournalState::Applying.as_str(),
})?;
Ok(HostCopy { run, journal })
}
async fn persist_created(
loaded: &LoadedProject,
writer: Option<&dyn ProjectRunWriter>,
writes: &mut TagWrites,
name: &str,
) -> Result<(), CoreError> {
let TagWrites { host, created, .. } = writes;
let (Some(writer), Some(host), Some(created)) = (writer, host.as_mut(), created.get(name))
else {
return Ok(());
};
if let Some(entry) = host.run.providers.iter_mut().find(|p| p.name == name) {
entry.created = Some(CreatedProof {
confirmed: created.confirmed,
signature: created.signature.clone(),
enabled: created.sent.0,
description: created.sent.1.clone(),
config: created.sent.2.clone(),
});
}
writer
.write_pre_image(
&loaded.target.gateway,
&loaded.file.project,
&host.run.run_id,
"run.json",
&host.run.to_bytes(),
)
.await
}
async fn persist_deleted(
host: Option<(&dyn ProjectRunWriter, &LoadedProject)>,
writes: &mut TagWrites,
name: &str,
paths: &[String],
) -> Result<(), CoreError> {
let (Some((writer, loaded)), Some(copy)) = (host, writes.host.as_mut()) else {
return Ok(());
};
let Some(entry) = copy.run.providers.iter_mut().find(|p| p.name == name) else {
return Err(CoreError::Internal(format!(
"the host pre-image lists no tag provider {name}"
)));
};
entry.deleted.extend(paths.iter().cloned());
entry.deleted.sort_by(|a, b| a.as_bytes().cmp(b.as_bytes()));
entry.deleted.dedup();
writer
.write_pre_image(
&loaded.target.gateway,
&loaded.file.project,
©.run.run_id,
"run.json",
©.run.to_bytes(),
)
.await
}
async fn persist_sent(
host: Option<(&dyn ProjectRunWriter, &LoadedProject)>,
writes: &mut TagWrites,
name: &str,
paths: &[String],
collided: &[String],
) -> Result<(), CoreError> {
let (Some((writer, loaded)), Some(copy)) = (host, writes.host.as_mut()) else {
return Ok(());
};
let Some(entry) = copy.run.providers.iter_mut().find(|p| p.name == name) else {
return Err(CoreError::Internal(format!(
"the host pre-image lists no tag provider {name}"
)));
};
for (list, more) in [
(entry.sent.get_or_insert_with(Vec::new), paths),
(&mut entry.collided, collided),
] {
list.extend(more.iter().cloned());
list.sort_by(|a, b| a.as_bytes().cmp(b.as_bytes()));
list.dedup();
}
writer
.write_pre_image(
&loaded.target.gateway,
&loaded.file.project,
©.run.run_id,
"run.json",
©.run.to_bytes(),
)
.await
}
fn host_journal_location(loaded: &LoadedProject) -> String {
project_journal_location(
LedgerRoot::for_node(&loaded.target.nodes.master),
&loaded.target.gateway,
&loaded.file.project,
)
}
async fn clear_host_journal(
result: &mut DeployProjectsResult,
loaded: &LoadedProject,
writer: Option<&dyn ProjectRunWriter>,
host: Option<&HostCopy>,
events: &mut RunEvents<'_>,
) {
let (Some(writer), Some(_)) = (writer, host) else {
return;
};
match writer
.clear_journal(&loaded.target.gateway, &loaded.file.project)
.await
{
Ok(()) => events.emit_late(ProjectsEvent::JournalCleared),
Err(err) => result.warnings.push(format!(
"the host journal {} could not be cleared ({}); the next run of this project refuses \
until it is removed",
host_journal_location(loaded),
err.code()
)),
}
}
async fn unwritten(
err: CoreError,
loaded: &LoadedProject,
writer: Option<&dyn ProjectRunWriter>,
host: Option<&HostCopy>,
events: &mut RunEvents<'_>,
) -> CoreError {
if let (Some(writer), Some(_)) = (writer, host)
&& writer
.clear_journal(&loaded.target.gateway, &loaded.file.project)
.await
.is_ok()
{
events.emit_late(ProjectsEvent::JournalCleared);
}
err
}
async fn journal_rolling_back(
result: &mut DeployProjectsResult,
loaded: &LoadedProject,
writer: Option<&dyn ProjectRunWriter>,
host: Option<&mut HostCopy>,
events: &mut RunEvents<'_>,
) {
let (Some(writer), Some(host)) = (writer, host) else {
return;
};
if host.journal.state == JournalState::RollingBack {
return;
}
host.journal.state = JournalState::RollingBack;
host.journal.started_at = rfc3339_now_utc();
match writer
.write_journal(
&loaded.target.gateway,
&loaded.file.project,
&host.journal.to_bytes(),
)
.await
{
Ok(()) => events.emit_late(ProjectsEvent::JournalWritten {
state: JournalState::RollingBack.as_str(),
}),
Err(err) => result.warnings.push(format!(
"the host journal {} could not be switched to rolling_back ({}); it still names this \
run",
host_journal_location(loaded),
err.code()
)),
}
}
impl TagWrites {
fn wrote(&self) -> bool {
!self.sent.is_empty()
}
}
struct CreatedProvider {
confirmed: bool,
sent: (bool, Option<String>, Value),
signature: Option<String>,
}
const READ_ATTEMPTS: usize = 3;
const READ_PAUSE: std::time::Duration = std::time::Duration::from_secs(2);
fn read_transient(err: &CoreError) -> bool {
match err {
CoreError::Internal(_) | CoreError::GatewayRestarting { .. } => true,
CoreError::Network {
source: Some(source),
..
} => source.is_timeout(),
_ => false,
}
}
async fn retry_transient<T, F, Fut>(mut read: F) -> Result<T, CoreError>
where
F: FnMut() -> Fut,
Fut: std::future::Future<Output = Result<T, CoreError>>,
{
let mut attempt = 1;
loop {
match read().await {
Err(err) if attempt < READ_ATTEMPTS && read_transient(&err) => {
attempt += 1;
tokio::time::sleep(READ_PAUSE).await;
}
answer => return answer,
}
}
}
fn holds_sent(record: &TagProviderRecord, sent: &(bool, Option<String>, Value)) -> bool {
record.enabled == sent.0
&& record.description() == sent.1.as_deref()
&& canonical(&record.config) == canonical(&sent.2)
}
struct TagFailure {
cause: RollbackCause,
what: String,
writes: TagWrites,
}
enum TagStepError {
Unwritten(CoreError),
Written(Box<TagFailure>),
}
impl From<CoreError> for TagStepError {
fn from(err: CoreError) -> Self {
TagStepError::Unwritten(err)
}
}
enum Stop {
Error(CoreError),
Failed(RollbackCause, String),
}
impl From<CoreError> for Stop {
fn from(err: CoreError) -> Self {
Stop::Error(err)
}
}
#[allow(clippy::too_many_arguments)]
async fn deploy_owned_tags(
plan: &DeployProjectsResult,
tag_records: &BTreeMap<String, TagRecord>,
loaded: &LoadedProject,
input: &ProjectsInput,
surfaces: &ProjectsSurfaces<'_>,
opts: &ProjectsRunOptions<'_>,
run_dir: &RunDir,
run_id: &RunId,
host: HostWrites<'_>,
route: &RouteState<'_>,
events: &mut RunEvents<'_>,
) -> Result<TagStep, TagStepError> {
let gateway = surfaces.gateway;
let Some(report) = &plan.tags else {
return Ok(TagStep {
carried: Vec::new(),
verify: None,
batches_sent: 0,
other_writes: 0,
writes: TagWrites::default(),
deleted: Vec::new(),
absent: Vec::new(),
});
};
let settings = fresh_provider_settings(loaded, gateway, report, run_dir, run_id).await?;
let creating: BTreeSet<&str> = report
.provider_settings
.iter()
.filter(|p| p.action == ProviderSettingsAction::Create)
.map(|p| p.provider.as_str())
.collect();
let mut live = BTreeMap::new();
let mut raw: BTreeMap<String, Vec<u8>> = BTreeMap::new();
for provider in &loaded.owned_tags {
if creating.contains(provider.name.as_str()) {
continue;
}
let bytes = gateway.tag_export(&provider.name).await?;
let mut file = run_dir
.create_private_file_nested(&["export", "tags"], &format!("{}.json", provider.name))?;
file.write_all(&bytes)
.and_then(|()| file.sync_all())
.map_err(|e| CoreError::DeployRunInvalid {
run_id: run_id.as_str().to_string(),
problem: format!(
"could not write the tag pre-image of provider {} ({:?})",
provider.name,
e.kind()
),
})?;
events.emit(ProjectsEvent::TagsPreImageExported {
provider: provider.name.clone(),
bytes: bytes.len(),
})?;
live.insert(
provider.name.clone(),
flatten_export(&bytes, &provider.name)?,
);
raw.insert(provider.name.clone(), bytes);
}
let now = plan_tags(&loaded.owned_tags, &live, tag_records);
if now.live_sha256 != report.live_sha256 {
return Err(CoreError::PlanStale {
expected: report.live_sha256.clone(),
actual: now.live_sha256,
detail: "the owned tag providers changed since the plan".to_string(),
}
.into());
}
let groups = fresh_groups(loaded, input, surfaces, report, &now, tag_records, &live).await?;
let empty = BTreeMap::new();
let owned = sorted(&loaded.owned_tags);
let mut queue: VecDeque<ImportBatch> = VecDeque::new();
let mut over_cap = Vec::new();
for (provider_plan, provider) in now.providers.iter().zip(&owned) {
let nodes = live.get(&provider.name).unwrap_or(&empty);
queue.extend(import_batches(
provider_plan,
provider,
nodes,
REMOVED_PROPERTY_POLICY,
)?);
over_cap.extend(rollback_over_cap(provider_plan, provider, nodes));
}
if !over_cap.is_empty() {
return Err(CoreError::deploy_findings(over_cap).into());
}
let mut refused = Vec::new();
for batch in queue.iter().filter(|batch| batch.carry) {
let nodes = live.get(&batch.provider).unwrap_or(&empty);
for (path, found) in uncarriable_in_batch(batch, nodes)? {
let file = owned
.iter()
.find(|p| p.name == batch.provider)
.and_then(|p| p.nodes.get(&path))
.map_or_else(
|| full_tag_path(&batch.provider, &path),
|node| node.file.clone(),
);
refused.push(DeployFinding {
file,
line: None,
column: None,
message: format!(
"{} hold(s) a live value and a valueSource that names no source (not a \
string): the repo drops a property, so the tag would go Overwrite, which \
cannot carry that value and would reset it; nothing was written",
found
.iter()
.map(|p| full_tag_path(&batch.provider, p))
.collect::<Vec<_>>()
.join(", ")
),
});
}
}
if !refused.is_empty() {
return Err(CoreError::deploy_findings(refused).into());
}
let mut writes = TagWrites::default();
if let Some(writer) = host.writer {
let release = surfaces
.host
.read_release_record(&loaded.target.gateway, &loaded.file.project)
.await?;
let pre = PreImageInputs {
project_zip: host.project_zip,
release: release.as_deref(),
exports: &raw,
settings: &settings,
records: tag_records,
plan: &now,
groups: groups.as_ref().is_some_and(GroupWork::would_change),
};
writes.host = Some(write_host_copy(writer, plan, loaded, run_id, &pre, events).await?);
}
drop(raw);
let outcome = send_owned_tags(
loaded,
surfaces,
opts,
run_id,
report,
&settings,
groups.as_ref(),
&now,
&live,
queue,
&mut writes,
host.writer,
route,
events,
)
.await;
writes.plan = Some(now);
writes.pre = live;
match outcome {
Ok((verify, other_writes)) => {
let mut carried = writes.carried.clone();
carried.sort_by(|a, b| a.as_bytes().cmp(b.as_bytes()));
let mut deleted: Vec<String> = writes
.applied
.iter()
.flat_map(|(name, applied)| applied.deleted.iter().map(|p| full_tag_path(name, p)))
.collect();
deleted.sort_by(|a, b| a.as_bytes().cmp(b.as_bytes()));
let absent = writes.delete_absent.clone();
Ok(TagStep {
carried,
verify: Some(verify),
batches_sent: writes.sent.len().saturating_sub(other_writes),
other_writes,
writes,
deleted,
absent,
})
}
Err(Stop::Error(err)) if !writes.wrote() => {
Err(TagStepError::Unwritten(
unwritten(err, loaded, host.writer, writes.host.as_ref(), events).await,
))
}
Err(Stop::Error(err)) => Err(TagStepError::Written(Box::new(TagFailure {
cause: RollbackCause::TagStepFailed,
what: format!("the tag step stopped ({})", err.code()),
writes,
}))),
Err(Stop::Failed(cause, what)) => Err(TagStepError::Written(Box::new(TagFailure {
cause,
what,
writes,
}))),
}
}
#[allow(clippy::too_many_arguments)]
async fn send_owned_tags(
loaded: &LoadedProject,
surfaces: &ProjectsSurfaces<'_>,
opts: &ProjectsRunOptions<'_>,
run_id: &RunId,
report: &TagsReport,
settings: &[(ProviderSettingsPlan, Option<TagProviderRecord>)],
groups: Option<&GroupWork>,
now: &TagPlan,
live: &ProviderReads,
mut queue: VecDeque<ImportBatch>,
writes: &mut TagWrites,
run_writer: Option<&dyn ProjectRunWriter>,
route: &RouteState<'_>,
events: &mut RunEvents<'_>,
) -> Result<(TagVerify, usize), Stop> {
let gateway = surfaces.gateway;
let failed = RollbackCause::TagStepFailed;
let repos = settings
.iter()
.map(|(planned, _)| provider_settings_of(loaded, &planned.provider))
.collect::<Result<Vec<_>, _>>()?;
let mut other_writes = 0;
for ((planned, record), repo) in settings.iter().zip(repos) {
if planned.action == ProviderSettingsAction::Same {
continue;
}
let name = planned.provider.as_str();
let answer = match (planned.action, record) {
(ProviderSettingsAction::Create, _) => {
let sent = (repo.enabled, repo.description.clone(), repo.config.clone());
let answer = gateway
.tag_provider_create(&[TagProviderCreate::with_settings(
name,
repo.enabled,
repo.description.clone(),
repo.config.clone(),
)])
.await;
match &answer {
Ok(()) => {
writes.created.insert(
name.to_string(),
CreatedProvider {
confirmed: true,
sent,
signature: None,
},
);
let _ = persist_created(loaded, run_writer, writes, name).await;
}
Err(err) if import_outcome_unknown(err) => {
let signature =
match retry_transient(|| gateway.tag_provider_record(name)).await {
Ok(Some(record)) if holds_sent(&record, &sent) => record.signature,
_ => None,
};
writes.created.insert(
name.to_string(),
CreatedProvider {
confirmed: false,
sent,
signature,
},
);
let _ = persist_created(loaded, run_writer, writes, name).await;
}
Err(_) => {}
}
answer
}
(ProviderSettingsAction::Modify, Some(record)) => {
writes.modified.push((name.to_string(), record.clone()));
let signature = record.signature.clone().unwrap_or_default();
gateway
.tag_provider_modify(&[TagProviderModify {
name: name.to_string(),
collection: record.collection().unwrap_or("core").to_string(),
enabled: repo.enabled,
description: repo.description.clone(),
config: repo.config.clone(),
signature,
}])
.await
}
_ => {
return Err(Stop::Error(CoreError::Internal(format!(
"tag provider {name} is planned to be modified but has no record"
))));
}
};
writes.sent.push(format!(
"[{name}] provider settings ({})",
planned.action.as_str()
));
other_writes += 1;
if let Err(err) = answer {
return Err(Stop::Failed(
failed,
format!(
"the settings of tag provider {name} were not applied ({})",
err.code()
),
));
}
let reread = retry_transient(|| gateway.tag_provider_record(name))
.await
.map_err(|err| {
Stop::Failed(
failed,
format!(
"tag provider {name} could not be read after its settings were applied ({})",
err.code()
),
)
})?;
if let Some(created) = writes.created.get_mut(name) {
created.signature = reread.as_ref().and_then(|r| r.signature.clone());
if let Err(err) = persist_created(loaded, run_writer, writes, name).await {
return Err(Stop::Failed(
failed,
format!(
"the host pre-image could not record tag provider {name}, which this run \
created ({})",
err.code()
),
));
}
}
let check = plan_provider_settings(name, repo, reread.as_ref());
if check.action != ProviderSettingsAction::Same {
return Err(Stop::Failed(
failed,
format!(
"the settings of tag provider {name} read back different from the repo ({})",
if check.properties.is_empty() {
"the provider is absent".to_string()
} else {
check.properties.join(", ")
}
),
));
}
events.emit_late(ProjectsEvent::TagProviderSettingsApplied {
provider: name.to_string(),
action: planned.action.as_str(),
});
}
if let Some(work) = groups.filter(|work| work.would_change()) {
let providers = work
.plans
.iter()
.filter(|p| p.would_change())
.map(|p| p.provider.clone())
.collect::<Vec<_>>();
let label = format!("tag groups of {}", providers.join(", "));
match commit_groups(loaded, surfaces, run_id, &report.plan_sha256, work).await {
Ok((written, deleted, record)) => {
writes
.sent
.push(format!("{label} ({written} written, {deleted} deleted)"));
other_writes += 1;
writes.groups = Some(GroupsDone::Committed {
record: Box::new(record),
providers: providers.clone(),
});
events.emit_late(ProjectsEvent::TagGroupsWritten {
providers,
written,
deleted,
});
}
Err(GroupFailure::NotWritten(err)) if !writes.wrote() => return Err(err.into()),
Err(GroupFailure::NotWritten(err)) => {
return Err(Stop::Failed(
failed,
format!("the {label} were not written ({})", err.code()),
));
}
Err(GroupFailure::Stale) if !writes.wrote() => {
return Err(CoreError::PlanStale {
expected: "the planned tag-group units".to_string(),
actual: "a unit changed before the commit".to_string(),
detail: "the owned providers' tag-group units changed since the plan"
.to_string(),
}
.into());
}
Err(GroupFailure::Stale) => {
return Err(Stop::Failed(
failed,
format!("the {label} were not written: a unit changed since the plan"),
));
}
Err(GroupFailure::Restored(why)) => {
writes.sent.push(format!("{label} (restored)"));
writes.groups = Some(GroupsDone::Restored);
return Err(Stop::Failed(
failed,
format!(
"the {label} did not verify ({why}); the group units were restored from \
the run's snapshot"
),
));
}
Err(GroupFailure::Unrestored(why)) => {
writes.sent.push(format!("{label} (not restored)"));
let journal = journal_location(
LedgerRoot::for_node(&loaded.target.nodes.master),
&loaded.target.gateway,
&loaded.target.mode,
);
writes.groups = Some(GroupsDone::Unrestored(format!(
"the {label} failed and could not be restored ({why}); the on-box journal \
{journal} stays"
)));
return Err(Stop::Failed(
failed,
format!("the {label} failed and could not be restored ({why})"),
));
}
}
}
while let Some(batch) = queue.pop_front() {
let name = batch.provider.clone();
let body = if batch.carry {
let fresh = match read_flat(gateway, &name).await {
Ok(fresh) => fresh,
Err(err) if !writes.wrote() => return Err(err.into()),
Err(err) => {
return Err(Stop::Failed(
failed,
format!(
"provider {name} could not be read to carry the live values ({})",
err.code()
),
));
}
};
let refused = uncarriable_in_batch(&batch, &fresh)?;
if !refused.is_empty() {
let paths: Vec<String> = refused
.iter()
.flat_map(|(_, found)| found.iter().map(|p| full_tag_path(&name, p)))
.collect();
let what = format!(
"{} hold(s) a live value and a valueSource that names no source in the \
export read right before their Overwrite, which would reset it, so the \
batch was not sent",
paths.join(", ")
);
if !writes.wrote() {
return Err(CoreError::deploy_tree_invalid(
full_tag_path(&name, &batch.path),
None,
None,
what,
)
.into());
}
return Err(Stop::Failed(failed, what));
}
match carried(&batch, &fresh)? {
Carried::Body(bytes) => bytes,
Carried::Split(parts) => {
for part in parts.into_iter().rev() {
queue.push_front(part);
}
continue;
}
}
} else {
batch.body.clone()
};
if batch.carry {
let event = ProjectsEvent::TagValuesCarried {
provider: name.clone(),
path: batch.path.clone(),
nodes: batch.nodes.len(),
};
if writes.wrote() {
events.emit_late(event);
} else {
events.emit(event)?;
}
writes
.carried
.extend(batch.nodes.iter().map(|p| full_tag_path(&name, p)));
}
let event = ProjectsEvent::TagImportSent {
provider: name.clone(),
path: batch.path.clone(),
policy: batch.policy.as_wire(),
nodes: batch.nodes.len(),
carry: batch.carry,
};
if writes.wrote() {
events.emit_late(event);
} else {
events.emit(event)?;
}
let host = run_writer.map(|writer| (writer, loaded));
if let Err(err) = persist_sent(host, writes, &name, &batch.nodes, &[]).await {
if !writes.wrote() {
return Err(Stop::Error(err));
}
return Err(Stop::Failed(
failed,
format!(
"the host pre-image could not record the tag import into provider {name} \
({}), so it was not sent",
err.code()
),
));
}
let answer = gateway
.tag_import(&name, &batch.path, batch.policy, body)
.await;
let sent_nodes = sent_bodies(loaded, now, &name, &batch);
let applied = writes.applied.entry(name.clone()).or_default();
applied.sent(&batch);
applied.bodies.extend(sent_nodes);
writes.sent.push(format!(
"{} ({}, {} node(s))",
full_tag_path(&name, &batch.path),
batch.policy.as_wire(),
batch.nodes.len()
));
let answer = answer.map_err(|err| {
Stop::Failed(
failed,
format!(
"the tag import into provider {name} failed ({})",
err.code()
),
)
})?;
let clean = answer.is_clean();
events.emit_late(ProjectsEvent::TagImportAnswered {
provider: name.clone(),
clean,
});
let collided = collided_paths(&name, &batch, &answer);
if !collided.is_empty() {
if let Some(applied) = writes.applied.get_mut(&name) {
applied.collided.extend(collided.iter().cloned());
}
let _ = persist_sent(host, writes, &name, &[], &collided).await;
}
if !clean {
let mut codes = answer.bad_quality.clone();
codes.truncate(TAG_FAILURE_PATHS_SHOWN);
let first = batch.nodes.first().map_or_else(
|| full_tag_path(&name, &batch.path),
|p| full_tag_path(&name, p),
);
return Err(Stop::Failed(
failed,
format!(
"the tag import headed by {first} into provider {name} was not clean ({} failure(s){})",
answer.failure_count.unwrap_or(codes.len() as u64),
if codes.is_empty() {
String::new()
} else {
format!(": {}", codes.join(", "))
}
),
));
}
}
let mut absent = Vec::new();
let host = run_writer.map(|writer| (writer, loaded));
let sent = delete_removed(gateway, now, route, writes, host, &mut absent, events).await;
writes.delete_absent = absent;
other_writes += sent?;
let owned = sorted(&loaded.owned_tags);
let names = owned_names(loaded);
let wrote = writes.wrote();
let (last, reads, settled, elapsed_s) = settled_reads(
gateway,
&names,
opts.tag_verify_settle,
opts.tag_verify_poll,
)
.await
.map_err(|err| {
if wrote {
Stop::Failed(
RollbackCause::TagVerifyFailed,
format!(
"the owned tag providers could not be read back to verify ({})",
err.code()
),
)
} else {
Stop::Error(err)
}
})?;
let empty = BTreeMap::new();
let mut mismatches = Vec::new();
for (provider_plan, provider) in now.providers.iter().zip(&owned) {
let name = provider.name.as_str();
let after = last.get(name).unwrap_or(&empty);
let pre = live.get(name).unwrap_or(&empty);
let created: BTreeSet<&str> = provider_plan.added.iter().map(String::as_str).collect();
let mut differ = Vec::new();
for (path, node) in &provider.nodes {
match after.get(path) {
None => differ.push(format!("{} (absent)", full_tag_path(name, path))),
Some(live_node) => {
let properties = diff_properties(
&node.json,
live_node,
path,
CanonOpts {
strip_values: true,
created: created.contains(path.as_str()),
exports: false,
},
);
if !properties.is_empty() {
differ.push(format!(
"{} ({})",
full_tag_path(name, path),
properties.join(", ")
));
}
}
}
}
for path in &provider_plan.removed {
if after.contains_key(path) {
differ.push(format!("{} (present)", full_tag_path(name, path)));
}
}
for path in &provider_plan.drift {
let properties = match (pre.get(path), after.get(path)) {
(Some(before), Some(now)) => diff_properties(before, now, path, EXPORTS),
(Some(_), None) => vec!["absent".to_string()],
_ => Vec::new(),
};
if !properties.is_empty() {
differ.push(format!(
"{} (drift: {})",
full_tag_path(name, path),
properties.join(", ")
));
}
}
events.emit_late(ProjectsEvent::TagsVerified {
provider: name.to_string(),
equal: differ.is_empty(),
});
mismatches.extend(differ);
}
mismatches.sort_by(|a, b| a.as_bytes().cmp(b.as_bytes()));
let mismatches_count = mismatches.len();
let what = (mismatches_count > 0).then(|| {
let shown: Vec<&str> = mismatches
.iter()
.take(TAG_FAILURE_PATHS_SHOWN)
.map(String::as_str)
.collect();
format!(
"the tag verify found {mismatches_count} node(s) unlike the repo or their pre-image \
after {reads} read(s){}: {}",
if settled { "" } else { ", unsettled" },
shown.join("; ")
)
});
let mut capped = mismatches;
capped.truncate(LIST_CAP);
let verify = TagVerify {
equal: mismatches_count == 0,
settled,
reads,
elapsed_s,
mismatches: capped,
mismatches_count,
};
writes.verify = Some(verify.clone());
if let Some(what) = what {
if !wrote {
return Err(CoreError::PlanStale {
expected: "the owned tags as planned".to_string(),
actual: format!("{mismatches_count} node(s) unlike the repo"),
detail: what,
}
.into());
}
return Err(Stop::Failed(RollbackCause::TagVerifyFailed, what));
}
Ok((verify, other_writes))
}
fn sent_bodies(
loaded: &LoadedProject,
now: &TagPlan,
provider: &str,
batch: &ImportBatch,
) -> Vec<(String, Value)> {
let Some(plan) = now.providers.iter().find(|p| p.provider == provider) else {
return Vec::new();
};
let Some(repo) = loaded.owned_tags.iter().find(|p| p.name == provider) else {
return Vec::new();
};
batch
.nodes
.iter()
.filter(|path| plan.added.binary_search(path).is_ok())
.filter_map(|path| {
repo.nodes
.get(path)
.map(|node| (path.clone(), node.json.clone()))
})
.collect()
}
fn collided_paths(provider: &str, batch: &ImportBatch, answer: &TagImportAnswer) -> Vec<String> {
let prefix = format!("[{provider}]");
let mut out = Vec::new();
let mut whole = answer.collisions_unnamed > 0;
for full in &answer.collided {
match full.strip_prefix(&prefix) {
Some(path) if batch.nodes.iter().any(|n| n == path) => out.push(path.to_string()),
_ => whole = true,
}
}
if whole {
return batch.nodes.clone();
}
out.sort_by(|a, b| a.as_bytes().cmp(b.as_bytes()));
out.dedup();
out
}
async fn delete_removed(
gateway: &dyn ProjectGateway,
now: &TagPlan,
route: &RouteState<'_>,
writes: &mut TagWrites,
host: Option<(&dyn ProjectRunWriter, &LoadedProject)>,
absent: &mut Vec<String>,
events: &mut RunEvents<'_>,
) -> Result<usize, Stop> {
let failed = RollbackCause::TagStepFailed;
let mut calls = 0;
for provider_plan in now.providers.iter().filter(|p| !p.removed.is_empty()) {
let name = provider_plan.provider.as_str();
let Some((client, handshake)) = route.ready() else {
return Err(Stop::Error(CoreError::Internal(format!(
"tag provider {name} has removals and the run has no ready tag-delete route"
))));
};
let fresh = match read_flat(gateway, name).await {
Ok(fresh) => fresh,
Err(err) if !writes.wrote() => return Err(err.into()),
Err(err) => {
return Err(Stop::Failed(
failed,
format!(
"provider {name} could not be read before its deletes ({})",
err.code()
),
));
}
};
absent.extend(
provider_plan
.removed
.iter()
.filter(|path| !fresh.contains_key(*path))
.map(|path| full_tag_path(name, path)),
);
let calls_of = match delete_calls(provider_plan, &fresh, handshake, TAG_DELETE_DESCENDANTS)
{
Ok(calls_of) => calls_of,
Err(refused) => {
let what = refused.describe(name);
if !writes.wrote() {
return Err(Stop::Error(CoreError::PlanStale {
expected: "the owned tags as planned".to_string(),
actual: "a removal no longer holds".to_string(),
detail: what,
}));
}
return Err(Stop::Failed(failed, what));
}
};
for chunk in calls_of.into_iter().map(|call| call.paths) {
if let Err(err) = persist_deleted(host, writes, name, &chunk).await {
let what = format!(
"the host pre-image could not record the deletes in provider {name} ({}), so \
they were not sent",
err.code()
);
if !writes.wrote() {
return Err(Stop::Error(err));
}
return Err(Stop::Failed(failed, what));
}
writes
.applied
.entry(name.to_string())
.or_default()
.deleted
.extend(chunk.iter().cloned());
writes
.sent
.push(format!("[{name}] tag delete ({} path(s))", chunk.len()));
calls += 1;
events.emit_late(ProjectsEvent::TagDeleteSent {
provider: name.to_string(),
paths: chunk.clone(),
rollback: false,
});
let answer = match client.delete(name, &chunk).await {
Ok(answer) => answer.check(name, &chunk).map(|()| answer),
Err(err) => Err(err),
};
let answer = answer.map_err(|err| {
Stop::Failed(
failed,
format!("the tag delete in provider {name} failed ({})", err.code()),
)
})?;
let clean = answer.is_clean();
let removed = answer.results.iter().filter(|r| r.removed()).count();
events.emit_late(ProjectsEvent::TagDeleteAnswered {
provider: name.to_string(),
clean,
deleted: removed,
});
if !clean {
let kept: Vec<String> = answer
.results
.iter()
.filter(|r| !r.removed())
.take(TAG_FAILURE_PATHS_SHOWN)
.map(|r| {
format!(
"{} ({})",
full_tag_path(name, &r.path),
r.not_removed_detail()
)
})
.collect();
return Err(Stop::Failed(
failed,
format!("the tag-delete route did not remove {}", kept.join(", ")),
));
}
}
}
Ok(calls)
}
struct ProjectUndo {
pre: Option<Vec<u8>>,
}
async fn project_sha(gateway: &dyn ProjectGateway, project: &str) -> Option<String> {
let zip = retry_transient(|| gateway.export_bytes(project))
.await
.ok()?;
live_digest(project, &zip).ok().map(|digest| digest.sha256)
}
fn settings_equal(a: &TagProviderRecord, b: &TagProviderRecord) -> bool {
a.enabled == b.enabled
&& a.description() == b.description()
&& canonical(&a.config) == canonical(&b.config)
}
async fn restore_provider_settings(
gateway: &dyn ProjectGateway,
name: &str,
pre: &TagProviderRecord,
) -> Result<bool, String> {
let now = retry_transient(|| gateway.tag_provider_record(name))
.await
.map_err(|e| format!("it could not be read ({})", e.code()))?
.ok_or_else(|| "it is no longer listed".to_string())?;
if settings_equal(&now, pre) {
return Ok(false);
}
gateway
.tag_provider_modify(&[TagProviderModify {
name: name.to_string(),
collection: pre
.collection()
.or(now.collection())
.unwrap_or("core")
.to_string(),
enabled: pre.enabled,
description: pre.description().map(str::to_string),
config: pre.config.clone(),
signature: now.signature.clone().unwrap_or_default(),
}])
.await
.map_err(|e| format!("the modify back failed ({})", e.code()))?;
match retry_transient(|| gateway.tag_provider_record(name)).await {
Ok(Some(back)) if settings_equal(&back, pre) => Ok(true),
Ok(_) => Err("it does not read back as its pre-image".to_string()),
Err(e) => Err(format!("it could not be read back ({})", e.code())),
}
}
async fn delete_created_provider(
gateway: &dyn ProjectGateway,
name: &str,
created: &CreatedProvider,
) -> Result<bool, String> {
let Some(record) = retry_transient(|| gateway.tag_provider_record(name))
.await
.map_err(|e| format!("it could not be read ({})", e.code()))?
else {
return Ok(false);
};
let Some(signature) = created.signature.as_deref() else {
return Err(if created.confirmed {
"its record could not be read right after the create, so nothing proves it is this \
run's; it was left in place"
.to_string()
} else {
"the create's answer was lost and the provider read right after it did not hold \
what this run sent, so it may be another client's; it was left in place"
.to_string()
});
};
if record.signature.as_deref() != Some(signature) {
return Err(
"it changed since this run's create (its signature is not the one read right after \
the create), so it was left in place"
.to_string(),
);
}
if !created.confirmed && !holds_sent(&record, &created.sent) {
return Err(
"the create's answer was lost and the provider no longer holds what this run sent, \
so it was left in place"
.to_string(),
);
}
gateway
.tag_provider_delete(name, signature)
.await
.map_err(|e| format!("its delete failed ({})", e.code()))?;
match retry_transient(|| gateway.tag_provider_record(name)).await {
Ok(None) => Ok(true),
Ok(Some(_)) => Err("it is still listed after the delete".to_string()),
Err(e) => Err(format!("its delete could not be read back ({})", e.code())),
}
}
async fn delete_created(
route: &dyn TagDeleteRoute,
handshake: &RouteHandshake,
provider: &str,
paths: &[String],
events: &mut RunEvents<'_>,
) -> (Vec<String>, Vec<(String, String)>) {
let mut removed = Vec::new();
let mut kept = Vec::new();
let chunks = match delete_chunks(paths, handshake) {
Ok(chunks) => chunks,
Err(err) => {
let why = format!(
"created by this run; it was not sent to the tag-delete route ({})",
err.code()
);
kept.extend(
paths
.iter()
.map(|p| (full_tag_path(provider, p), why.clone())),
);
return (removed, kept);
}
};
for chunk in chunks.into_iter().filter(|chunk| !chunk.is_empty()) {
events.emit_late(ProjectsEvent::TagDeleteSent {
provider: provider.to_string(),
paths: chunk.clone(),
rollback: true,
});
let answer = match route.delete(provider, &chunk).await {
Ok(answer) => answer.check(provider, &chunk).map(|()| answer),
Err(err) => Err(err),
};
match answer {
Ok(answer) => {
let mut count = 0;
for result in &answer.results {
let full = full_tag_path(provider, &result.path);
if result.removed() {
count += 1;
removed.push(full);
} else if result.exists_after.is_none() {
kept.push((
full,
format!(
"created by this run; the tag-delete route could not confirm \
it was removed (quality {})",
result.quality.as_deref().unwrap_or("none")
),
));
} else {
kept.push((
full,
format!(
"created by this run; the tag-delete route did not remove it \
(quality {})",
result.quality.as_deref().unwrap_or("none")
),
));
}
}
events.emit_late(ProjectsEvent::TagDeleteAnswered {
provider: provider.to_string(),
clean: answer.is_clean(),
deleted: count,
});
}
Err(err) => {
let why = format!(
"created by this run; its delete call failed ({})",
err.code()
);
kept.extend(
chunk
.iter()
.map(|p| (full_tag_path(provider, p), why.clone())),
);
}
}
}
(removed, kept)
}
fn sorted_capped(mut items: Vec<String>) -> (Vec<String>, usize) {
items.sort_by(|a, b| a.as_bytes().cmp(b.as_bytes()));
items.dedup();
capped(items)
}
#[allow(clippy::too_many_arguments)]
async fn roll_back(
failure: TagFailure,
project_undo: Option<ProjectUndo>,
mut result: DeployProjectsResult,
loaded: &LoadedProject,
input: &ProjectsInput,
surfaces: &ProjectsSurfaces<'_>,
opts: &ProjectsRunOptions<'_>,
run_id: &RunId,
rel: &str,
undo_writers: UndoWriters<'_>,
route: &RouteState<'_>,
events: &mut RunEvents<'_>,
) -> Result<DeployProjectsResult, CoreError> {
let run_writer = undo_writers.run;
let gateway = surfaces.gateway;
let TagFailure {
cause,
what,
mut writes,
} = failure;
let project = result.project.clone();
let backup_since_ms = wall_clock_ms();
events.emit_late(ProjectsEvent::RollbackStarted { cause });
journal_rolling_back(
&mut result,
loaded,
run_writer,
writes.host.as_mut(),
events,
)
.await;
let mut left: Vec<String> = Vec::new();
let mut reimported: Vec<String> = Vec::new();
let mut carried_back: Vec<String> = Vec::new();
let mut undeletable: Vec<String> = Vec::new();
let mut undeletable_left: Vec<String> = Vec::new();
let mut deleted: Vec<String> = Vec::new();
let mut delete_left: Vec<(String, String)> = Vec::new();
let mut uncarriable_paths: Vec<String> = Vec::new();
let mut recreated_tops: BTreeMap<String, Vec<String>> = BTreeMap::new();
let mut project_reimported = false;
if let Some(undo) = &project_undo {
match &undo.pre {
Some(zip) => match live_digest(&project, zip) {
Ok(pre) => {
let now = project_sha(gateway, &project).await;
if now.as_deref() != Some(pre.sha256.as_str()) {
match gateway.project_import(&project, zip.clone(), true).await {
Ok(_) => project_reimported = true,
Err(err) => left.push(format!(
"project {project}: the import of its pre-image failed ({})",
err.code()
)),
}
let back = project_sha(gateway, &project).await;
let equal = back.as_deref() == Some(pre.sha256.as_str());
events.emit_late(ProjectsEvent::ProjectRestored { equal });
if !equal && project_reimported {
left.push(format!(
"project {project}: it does not read back as its pre-image \
({rel}/export/{project}.zip)"
));
}
}
}
Err(err) => left.push(format!(
"project {project}: its pre-image cannot be read ({})",
err.code()
)),
},
None => match gateway.project_find(&project).await {
Ok(Some(_)) => result.warnings.push(format!(
"project {project} was created by this run and stays: the gateway surface \
has no project delete, so the rollback cannot remove it"
)),
Ok(None) => {}
Err(err) => result.warnings.push(format!(
"project {project}: whether this run created it could not be read ({})",
err.code()
)),
},
}
}
let empty = BTreeMap::new();
for provider in sorted(&loaded.owned_tags) {
let name = provider.name.as_str();
if writes.created.contains_key(name) {
continue;
}
let Some(applied) = writes.applied.get(name).filter(|a| !a.is_empty()) else {
continue;
};
let Some(provider_plan) = writes
.plan
.as_ref()
.and_then(|p| p.providers.iter().find(|pp| pp.provider == name))
else {
left.push(format!("[{name}]: no plan to roll back against"));
continue;
};
let pre = writes.pre.get(name).unwrap_or(&empty);
let current = match read_flat(gateway, name).await {
Ok(current) => current,
Err(err) => {
left.push(format!(
"[{name}]: the provider could not be read to plan its rollback ({}); every \
node the run sent stays as it is",
err.code()
));
continue;
}
};
let no_record = BTreeSet::new();
let recreate = writes.recreate.get(name).unwrap_or(&no_record);
let undo = match rollback_plan(
provider_plan,
pre,
¤t,
applied,
false,
recreate,
route.ready().map(|(_, handshake)| handshake),
TAG_DELETE_DESCENDANTS,
) {
Ok(undo) => undo,
Err(err) => {
left.push(format!(
"[{name}]: its rollback could not be planned ({})",
err.code()
));
continue;
}
};
if !undo.recreated_deleted.is_empty() {
recreated_tops
.entry(name.to_string())
.or_default()
.extend(undo.recreated_deleted.iter().cloned());
}
let kept = route.kept_reason(name);
for path in undo
.undeletable
.iter()
.filter(|p| !undo.created.contains(p))
{
let full = full_tag_path(name, path);
undeletable_left.push(format!("{full} ({kept})"));
undeletable.push(full);
}
for (path, why) in &undo.foreign {
let full = full_tag_path(name, path);
undeletable_left.push(format!("{full} ({})", why.describe()));
undeletable.push(full);
}
uncarriable_paths.extend(undo.uncarriable.iter().map(|p| full_tag_path(name, p)));
let full = |batch: &ImportBatch| -> Vec<String> {
batch.nodes.iter().map(|p| full_tag_path(name, p)).collect()
};
let mut queue: VecDeque<ImportBatch> = undo.imports.into();
while let Some(batch) = queue.pop_front() {
let body = if batch.carry {
let fresh = match read_flat(gateway, name).await {
Ok(fresh) => fresh,
Err(err) => {
left.extend(full(&batch).into_iter().map(|p| {
format!(
"{p} (not restored: the export to carry its value from failed, {})",
err.code()
)
}));
continue;
}
};
match uncarriable_in_batch(&batch, &fresh) {
Ok(bad) if bad.is_empty() => {}
Ok(bad) => {
let bad: BTreeSet<String> = bad
.into_iter()
.map(|(p, _)| full_tag_path(name, &p))
.collect();
left.extend(full(&batch).into_iter().filter(|p| !bad.contains(p)).map(
|p| format!("{p} (not restored: its batch held an uncarriable tag)"),
));
uncarriable_paths.extend(bad);
continue;
}
Err(err) => {
left.extend(
full(&batch)
.into_iter()
.map(|p| format!("{p} (not restored: {})", err.code())),
);
continue;
}
}
match carried(&batch, &fresh) {
Ok(Carried::Body(bytes)) => bytes,
Ok(Carried::Split(parts)) => {
for part in parts.into_iter().rev() {
queue.push_front(part);
}
continue;
}
Err(err) => {
left.extend(
full(&batch)
.into_iter()
.map(|p| format!("{p} (not restored: {})", err.code())),
);
continue;
}
}
} else {
batch.body.clone()
};
if batch.carry {
events.emit_late(ProjectsEvent::TagValuesCarried {
provider: name.to_string(),
path: batch.path.clone(),
nodes: batch.nodes.len(),
});
}
events.emit_late(ProjectsEvent::TagImportSent {
provider: name.to_string(),
path: batch.path.clone(),
policy: batch.policy.as_wire(),
nodes: batch.nodes.len(),
carry: batch.carry,
});
let label = format!(
"{} ({}, {} node(s))",
full_tag_path(name, &batch.path),
batch.policy.as_wire(),
batch.nodes.len()
);
match gateway
.tag_import(name, &batch.path, batch.policy, body)
.await
{
Ok(answer) => {
let clean = answer.is_clean();
events.emit_late(ProjectsEvent::TagImportAnswered {
provider: name.to_string(),
clean,
});
if clean {
reimported.extend(full(&batch));
if batch.carry {
carried_back.extend(full(&batch));
}
} else {
left.push(format!("{label}: the restore import was not clean"));
}
}
Err(err) => left.push(format!(
"{label}: the restore import failed ({})",
err.code()
)),
}
}
if !undo.created.is_empty() {
let left_all = |why: String| -> Vec<(String, String)> {
undo.created
.iter()
.map(|p| (full_tag_path(name, p), why.clone()))
.collect()
};
let Some((client, handshake)) = route.ready() else {
delete_left.extend(left_all(
"created by this run; the tag-delete route is not ready".to_string(),
));
continue;
};
let fresh = match read_flat(gateway, name).await {
Ok(fresh) => fresh,
Err(err) => {
delete_left.extend(left_all(format!(
"created by this run; the provider could not be read before its deletes \
({})",
err.code()
)));
continue;
}
};
match created_deletes_now(name, &undo.created, &fresh, applied, TAG_DELETE_DESCENDANTS)
{
Ok(drawn) => {
let deletes = drawn.deletes;
let foreign = drawn.foreign;
for path in &drawn.kept {
let full = full_tag_path(name, path);
undeletable_left.push(format!("{full} ({kept})"));
undeletable.push(full);
}
for (path, user) in &drawn.in_use {
let full = full_tag_path(name, path);
undeletable_left.push(format!(
"{full} (created by this run; the live node {} still uses it, so it \
was not removed)",
full_tag_path(name, user)
));
undeletable.push(full);
}
for (path, why) in &foreign {
let full = full_tag_path(name, path);
undeletable_left.push(format!("{full} ({})", why.describe()));
undeletable.push(full);
}
if !deletes.is_empty() {
let (removed, kept) =
delete_created(client, handshake, name, &deletes, events).await;
deleted.extend(removed);
delete_left.extend(kept);
}
}
Err(err) => delete_left.extend(left_all(format!(
"created by this run; its deletes could not be planned ({})",
err.code()
))),
}
}
}
let mut groups_restored = false;
match &writes.groups {
Some(GroupsDone::Committed { record, providers }) => {
let journal = journal_location(
LedgerRoot::for_node(&loaded.target.nodes.master),
&loaded.target.gateway,
&loaded.target.mode,
);
match (surfaces.apply_host, surfaces.deploy) {
(Some(host), Some(deploy)) => {
match restore_groups(host, deploy, record, run_id, None).await {
Ok(()) => {
groups_restored = true;
events.emit_late(ProjectsEvent::TagGroupsRestored {
providers: providers.clone(),
});
}
Err(why) => left.push(format!(
"the tag groups of {}: {why}; see the on-box journal {journal}",
providers.join(", ")
)),
}
}
_ => left.push(format!(
"the tag groups of {}: no apply host to restore them with",
providers.join(", ")
)),
}
}
Some(GroupsDone::Restored) => groups_restored = true,
Some(GroupsDone::Unrestored(why)) => left.push(why.clone()),
None => {}
}
let mut provider_restored = Vec::new();
let mut provider_deleted = Vec::new();
for (name, pre) in writes.modified.iter().rev() {
match restore_provider_settings(gateway, name, pre).await {
Ok(true) => {
provider_restored.push(name.clone());
events.emit_late(ProjectsEvent::TagProviderRestored {
provider: name.clone(),
action: "modify",
});
}
Ok(false) => {}
Err(why) => left.push(format!(
"[{name}] provider settings: {why} (pre-image {rel}/export/tags/{name}.provider.json)"
)),
}
}
for (name, created) in &writes.created {
match delete_created_provider(gateway, name, created).await {
Ok(true) => {
provider_deleted.push(name.clone());
events.emit_late(ProjectsEvent::TagProviderRestored {
provider: name.clone(),
action: "delete",
});
}
Ok(false) => {}
Err(why) => left.push(format!("[{name}] the provider this run created: {why}")),
}
}
let owned = sorted(&loaded.owned_tags);
let names: Vec<&str> = owned
.iter()
.map(|p| p.name.as_str())
.filter(|name| writes.pre.contains_key(*name))
.collect();
let exempt: BTreeSet<String> = undeletable
.iter()
.chain(&uncarriable_paths)
.chain(delete_left.iter().map(|(path, _)| path))
.cloned()
.collect();
let mut verify = None;
if !names.is_empty() {
match settled_reads(
gateway,
&names,
opts.tag_verify_settle,
opts.tag_verify_poll,
)
.await
{
Ok((last, reads, settled, elapsed_s)) => {
let mut mismatches = Vec::new();
for name in &names {
let pre = writes.pre.get(*name).unwrap_or(&empty);
let now = last.get(*name).unwrap_or(&empty);
let tops = recreated_tops.get(*name).map(Vec::as_slice).unwrap_or(&[]);
for (path, before) in pre {
let full = full_tag_path(name, path);
if exempt.contains(&full) {
continue;
}
match now.get(path) {
None => mismatches.push(format!("{full} (absent)")),
Some(after) => {
let opts = if recreated_at(path, tops) {
RECREATED
} else {
EXPORTS
};
let properties = diff_properties(before, after, path, opts);
if !properties.is_empty() {
mismatches.push(format!("{full} ({})", properties.join(", ")));
}
}
}
}
for path in now.keys() {
let full = full_tag_path(name, path);
if !pre.contains_key(path) && !exempt.contains(&full) {
mismatches.push(format!("{full} (not in the pre-image)"));
}
}
}
mismatches.sort_by(|a, b| a.as_bytes().cmp(b.as_bytes()));
let equal = mismatches.is_empty();
events.emit_late(ProjectsEvent::RollbackVerified { equal });
left.extend(mismatches.iter().map(|m| format!("verify: {m}")));
let mismatches_count = mismatches.len();
mismatches.truncate(LIST_CAP);
verify = Some(TagVerify {
equal,
settled,
reads,
elapsed_s,
mismatches,
mismatches_count,
});
}
Err(err) => left.push(format!(
"the owned tag providers could not be read back to verify the rollback ({})",
err.code()
)),
}
}
left.extend(undeletable_left);
left.extend(
delete_left
.iter()
.map(|(path, why)| format!("{path} ({why})")),
);
left.extend(uncarriable_paths.iter().map(|p| {
format!("{p} (its live value cannot be carried into an Overwrite, so it was not restored)")
}));
if left.is_empty() {
result.outcome = if project_undo.is_some() {
ProjectOutcome::RolledBack
} else {
ProjectOutcome::TagsRolledBack
};
result.ok = false;
result.slug = Some(ROLLED_BACK_SLUG.to_string());
result.record_written = false;
result.restore_command = None;
if let Some(tags) = result.tags.as_mut() {
let (carried, carried_count) = sorted_capped(writes.carried.clone());
tags.carried = carried;
tags.carried_count = carried_count;
tags.verify = writes.verify.clone();
let (reimported, reimported_count) = sorted_capped(reimported);
let (carried, carried_count) = sorted_capped(carried_back);
provider_restored.sort_by(|a, b| a.as_bytes().cmp(b.as_bytes()));
provider_deleted.sort_by(|a, b| a.as_bytes().cmp(b.as_bytes()));
let (deleted, deleted_count) = sorted_capped(deleted);
tags.rollback = Some(TagRollback {
cause: cause.as_str().to_string(),
detail: what,
reimported,
reimported_count,
carried,
carried_count,
undeletable: Vec::new(),
undeletable_count: 0,
groups_restored,
provider_restored,
provider_deleted,
project_reimported,
verify,
deleted,
deleted_count,
});
}
let records_back = match &writes.records_back {
Some(back) => {
restore_tag_records(&mut result, loaded, undo_writers.records, back, events).await
}
None => true,
};
if let Some(back) = &writes.release_back {
restore_release_record(&mut result, loaded, undo_writers.release, back.as_deref())
.await;
}
if records_back {
clear_host_journal(
&mut result,
loaded,
run_writer,
writes.host.as_ref(),
events,
)
.await;
}
if backup_step(&mut result, &names, surfaces, opts, backup_since_ms, events).await
== Some(false)
{
let reason = result
.tags
.as_ref()
.and_then(|tags| tags.backup.as_ref())
.and_then(|backup| backup.reason)
.map_or("unknown", |reason| reason.as_str());
result.warnings.push(format!(
"the backup did not converge after the rollback ({reason}): the master is \
restored and verified, and data.tags.backup says what the backup holds"
));
}
events.emit_late(ProjectsEvent::Finished {
outcome: result.outcome,
});
note_late_log_failure(&mut result, events);
return Ok(result);
}
let listed = |items: &[String]| {
let shown: Vec<&str> = items
.iter()
.take(TAG_FAILURE_PATHS_SHOWN)
.map(String::as_str)
.collect();
let more = items.len().saturating_sub(shown.len());
format!(
"{}{}",
if shown.is_empty() {
"none".to_string()
} else {
shown.join("; ")
},
if more > 0 {
format!(" and {more} more")
} else {
String::new()
}
)
};
let pre_images = match (&writes.host, cause) {
(Some(_), RollbackCause::RollbackRun) => pre_image_location(
LedgerRoot::for_node(&loaded.target.nodes.master),
&loaded.target.gateway,
&loaded.file.project,
crate::deploy::project_run::PRE_IMAGE_DIR,
),
_ => format!("{rel}/export"),
};
Err(CoreError::RollbackFailed(Box::new(RollbackFailure {
run_id: run_id.as_str().to_string(),
env: input.env.clone(),
run_dir: Some(rel.to_string()),
journal: pre_images,
cause: format!(
"{}: {what}; {} {} tag write(s): {}; the rollback put back what it could \
and {} item(s) are left: {}{}",
cause.as_str(),
if cause == RollbackCause::RollbackRun {
"the undone run planned"
} else {
"the run sent"
},
writes.sent.len(),
listed(&writes.sent),
left.len(),
listed(&left),
match &writes.host {
Some(host) => format!(
"; the host journal {} stays, and `{}` retries the undo from the host copy",
host_journal_location(loaded),
rollback_run_command(&input.env, &host.journal.run_id)
),
None => String::new(),
}
),
scope: RollbackScope::Projects,
})))
}
pub fn rollback_run_prose(
loaded: &LoadedProject,
env: &str,
run_id: &str,
route: &TagDeleteSurface<'_>,
) -> String {
let created = match route {
TagDeleteSurface::Configured(_) => {
"Tags the run created in a provider that existed are removed through the tag-delete \
route when it answers and lists the provider; one the route cannot remove, or that \
is not what the run wrote, stays and ends the undo as rollback_failed"
}
TagDeleteSurface::NotConfigured { .. } => {
"Tags the run created in a provider that existed cannot be removed without the \
tag-delete route, which this profile does not configure, and end the undo as \
rollback_failed"
}
};
format!(
"deploy projects will undo run {run_id} of project {} on gateway {} (env {env}) from \
the copy the gateway host keeps: the owned tags of {}, their tag groups, provider \
settings and tag records, and the project, back to that run's pre-image. Tags the run \
deleted are re-created from the pre-image. {created}. `ign deploy projects --env {env} \
--check --rollback-run {run_id}` lists what it restores",
loaded.file.project,
loaded.target.gateway,
loaded.file.tag_providers.join(", "),
)
}
struct HostSet {
run: PreImageRun,
journal: Option<ProjectJournal>,
project_zip: Option<Vec<u8>>,
release: Option<Vec<u8>>,
pre: ProviderReads,
settings: BTreeMap<String, TagProviderRecord>,
records: BTreeMap<String, (TagRecord, Vec<u8>)>,
sent: BTreeMap<String, BTreeMap<String, Value>>,
}
async fn read_host_set(
loaded: &LoadedProject,
input: &ProjectsInput,
host: &dyn ProjectHost,
target: &RunId,
) -> Result<HostSet, CoreError> {
let gateway = &loaded.target.gateway;
let project = &loaded.file.project;
let root = LedgerRoot::for_node(&loaded.target.nodes.master);
let invalid = |problem: String| CoreError::DeployRunInvalid {
run_id: target.as_str().to_string(),
problem,
};
let journal = read_project_journal(loaded, host).await?;
if let Some(journal) = &journal
&& journal.run_id != target.as_str()
{
return Err(project_journal_present(loaded, &input.env, &journal.run_id));
}
let run_location = pre_image_location(root, gateway, project, "run.json");
let Some(bytes) = host.read_pre_image(gateway, project, "run.json").await? else {
return Err(invalid(format!(
"{run_location} does not exist: the host keeps the pre-image of the latest \
tag-owning run of this project only, and holds none"
)));
};
let run = PreImageRun::parse(&bytes, &run_location, gateway, project)?;
if run.run_id != target.as_str() {
return Err(invalid(format!(
"the host keeps the pre-image of run {} only, the latest run of this project; \
run {} cannot be undone from it",
run.run_id,
target.as_str()
)));
}
for provider in &run.providers {
if !loaded.file.tag_providers.contains(&provider.name) {
return Err(invalid(format!(
"{run_location} lists tag provider {}, which deploy/project.yaml does not own: \
only the project's own providers are restored",
provider.name
)));
}
}
if run.project_action == "import" && !run.files.contains_key("project.zip") {
return Err(invalid(format!(
"{run_location}: the run imported the project but the set holds no project.zip"
)));
}
let mut set = HostSet {
run,
journal,
project_zip: None,
release: None,
pre: BTreeMap::new(),
settings: BTreeMap::new(),
records: BTreeMap::new(),
sent: BTreeMap::new(),
};
for (name, sha) in &set.run.files {
let location = pre_image_location(root, gateway, project, name);
let Some(bytes) = host.read_pre_image(gateway, project, name).await? else {
return Err(invalid(format!(
"{location} is missing: the pre-image set is incomplete"
)));
};
if sha256_hex(&bytes) != *sha {
return Err(invalid(format!(
"{location} does not hash as run.json records: the set is incomplete or changed"
)));
}
match PreImageFile::parse(name) {
Some(PreImageFile::Project) => set.project_zip = Some(bytes),
Some(PreImageFile::Release) => {
ReleaseRecord::parse(&bytes, &location, gateway, project)?;
set.release = Some(bytes);
}
Some(PreImageFile::Tags(provider)) => {
let flat = flatten_export(&bytes, &provider)?;
set.pre.insert(provider, flat);
}
Some(PreImageFile::Provider(provider)) => {
let record: TagProviderRecord = serde_json::from_slice(&bytes)
.map_err(|_| invalid(format!("{location} is not a tag provider record")))?;
set.settings.insert(provider, record);
}
Some(PreImageFile::Record(provider)) => {
let record = TagRecord::parse(&bytes, &location, gateway, &provider)?;
if record.project != *project {
return Err(invalid(format!(
"{location} is another project's tag record"
)));
}
set.records.insert(provider, (record, bytes));
}
Some(PreImageFile::Sent(provider)) => {
let sent: BTreeMap<String, Value> =
serde_json::from_slice(&bytes).map_err(|_| {
invalid(format!(
"{location} is not an object of node paths to the nodes the run sent"
))
})?;
set.sent.insert(provider, sent);
}
Some(PreImageFile::Run) | None => {
return Err(invalid(format!("{location}: not a file of the set")));
}
}
}
let empty = BTreeMap::new();
for provider in &set.run.providers {
let pre = set.pre.get(&provider.name).unwrap_or(&empty);
if let Some(path) = provider.deleted.iter().find(|p| !pre.contains_key(*p)) {
return Err(invalid(format!(
"{run_location} names {} among the paths run {} deleted, and the set's \
pre-image of the provider does not hold it: the set is stale or partial",
full_tag_path(&provider.name, path),
target.as_str()
)));
}
}
Ok(set)
}
fn undo_writes(set: &HostSet) -> TagWrites {
let mut writes = TagWrites::default();
let mut providers = Vec::new();
let mut records_back = Vec::new();
for provider in &set.run.providers {
let name = provider.name.clone();
providers.push(ProviderPlan {
provider: name.clone(),
added: provider.added.clone(),
recreated: Vec::new(),
changed: provider
.changed
.iter()
.map(|path| NodeChange {
path: path.clone(),
properties: Vec::new(),
restart_likely: false,
})
.collect(),
kept_values: Vec::new(),
removed: provider.removed.clone(),
already_absent: Vec::new(),
drift: Vec::new(),
same: 0,
first_change: false,
record_only: false,
provider_emptied: false,
record_run_id: None,
});
let mut applied = AppliedSteps::default();
match &provider.sent {
Some(sent) => applied.nodes.extend(sent.iter().cloned()),
None => {
applied.nodes.extend(provider.added.iter().cloned());
applied.nodes.extend(provider.changed.iter().cloned());
}
}
if let Some(bodies) = set.sent.get(&name) {
applied.bodies.extend(
bodies
.iter()
.filter(|(path, _)| applied.nodes.contains(*path))
.map(|(path, node)| (path.clone(), node.clone())),
);
}
applied.collided.extend(provider.collided.iter().cloned());
applied.deleted.clone_from(&provider.deleted);
if !applied.is_empty() {
writes.sent.push(format!(
"[{name}] {} planned node(s), {} deleted path(s)",
applied.nodes.len(),
applied.deleted.len()
));
writes.applied.insert(name.clone(), applied);
}
if let Some(proof) = &provider.created {
writes
.sent
.push(format!("[{name}] provider settings (create)"));
writes.created.insert(
name.clone(),
CreatedProvider {
confirmed: proof.confirmed,
sent: (
proof.enabled,
proof.description.clone(),
proof.config.clone(),
),
signature: proof.signature.clone(),
},
);
}
if provider.settings == "modify"
&& let Some(record) = set.settings.get(&name)
{
writes
.sent
.push(format!("[{name}] provider settings (modify)"));
writes.modified.push((name.clone(), record.clone()));
}
match set.records.get(&name) {
Some((record, bytes)) => {
writes
.recreate
.insert(name.clone(), record.nodes.iter().cloned().collect());
records_back.push((name, Some(bytes.clone())));
}
None => records_back.push((name, None)),
}
}
writes.plan = Some(TagPlan {
providers,
tags_sha256: String::new(),
live_sha256: String::new(),
plan_sha256: String::new(),
});
writes.pre = set.pre.clone();
writes.records_back = Some(records_back);
writes.release_back = Some(set.release.clone());
writes
}
async fn undo_groups(
set: &HostSet,
input: &ProjectsInput,
apply_host: Option<&dyn ApplyHost>,
target: &RunId,
) -> Option<GroupsDone> {
if !set.run.groups {
return None;
}
let providers: Vec<String> = set.run.providers.iter().map(|p| p.name.clone()).collect();
let recover = format!(
"the tag groups of {}: the group commit of run {} did not finish on the box; recover \
it with `ign deploy rollback --run {} --env {}` first",
providers.join(", "),
target.as_str(),
target.as_str(),
shell_word(&input.env)
);
let Some(host) = apply_host else {
return Some(GroupsDone::Unrestored(format!(
"the tag groups of {}: no apply host to restore them with",
providers.join(", ")
)));
};
match host.read_record(target).await {
Ok(None) => None,
Ok(Some(record)) => match record.state {
RecordState::RolledBack => None,
RecordState::Applied => match live_changes(host, &record).await {
Ok(changed) if changed.is_empty() => Some(GroupsDone::Committed {
record: Box::new(record),
providers,
}),
Ok(changed) => Some(GroupsDone::Unrestored(format!(
"the tag groups of {}: {} unit file(s) changed since run {} ({}{}); they were \
not put back over that change",
providers.join(", "),
changed.len(),
target.as_str(),
changed
.iter()
.take(APPLY_LIST_SHOWN)
.cloned()
.collect::<Vec<_>>()
.join(", "),
if changed.len() > APPLY_LIST_SHOWN {
", ..."
} else {
""
}
))),
Err(err) => Some(GroupsDone::Unrestored(format!(
"the tag groups of {}: their unit files could not be compared with run {} ({})",
providers.join(", "),
target.as_str(),
err.code()
))),
},
RecordState::Applying => Some(GroupsDone::Unrestored(recover)),
},
Err(err) => Some(GroupsDone::Unrestored(format!(
"the tag groups of {}: the run record of run {} could not be read ({})",
providers.join(", "),
target.as_str(),
err.code()
))),
}
}
async fn restore_release_record(
result: &mut DeployProjectsResult,
loaded: &LoadedProject,
writer: Option<&dyn ReleaseRecordWriter>,
back: Option<&[u8]>,
) {
let (Some(writer), Some(bytes)) = (writer, back) else {
result.warnings.push(format!(
"the release record of {} was not put back: the host copy holds none from before \
the run, so the record may still name the undone release",
loaded.file.project
));
return;
};
if let Err(err) = writer
.write_release_record(&loaded.target.gateway, &loaded.file.project, bytes)
.await
{
result.warnings.push(format!(
"the release record of {} could not be put back ({}); it may still name the \
undone release",
loaded.file.project,
err.code()
));
}
}
async fn restore_tag_records(
result: &mut DeployProjectsResult,
loaded: &LoadedProject,
writer: Option<&dyn TagRecordWriter>,
back: &[(String, Option<Vec<u8>>)],
events: &mut RunEvents<'_>,
) -> bool {
let mut ok = true;
for (provider, bytes) in back {
let Some(bytes) = bytes else {
result.warnings.push(format!(
"tag provider {provider} had no tag record before the undone run: the record \
that run left, if any, stays (this release cannot remove one)"
));
continue;
};
let Some(writer) = writer else {
ok = false;
continue;
};
match writer
.write_tag_record(&loaded.target.gateway, provider, bytes)
.await
{
Ok(()) => events.emit_late(ProjectsEvent::TagRecordWritten {
provider: provider.clone(),
}),
Err(err) => {
ok = false;
events.emit_late(ProjectsEvent::TagRecordFailed {
provider: provider.clone(),
code: err.code().to_string(),
});
result.warnings.push(format!(
"the tag record of {provider} could not be put back ({}); the host journal \
{} stays",
err.code(),
host_journal_location(loaded)
));
}
}
}
ok
}
fn action_of(word: &str) -> ProjectAction {
match word {
"create" => ProjectAction::Create,
"import" => ProjectAction::Import,
"record" => ProjectAction::Record,
_ => ProjectAction::Skip,
}
}
fn rollback_run_result(
loaded: &LoadedProject,
input: &ProjectsInput,
set: &HostSet,
writes: &TagWrites,
live_sha256: Option<String>,
check: bool,
) -> DeployProjectsResult {
let empty = TagPlan {
providers: Vec::new(),
tags_sha256: String::new(),
live_sha256: String::new(),
plan_sha256: String::new(),
};
let mut report = TagsReport::from_plan(
writes.plan.as_ref().unwrap_or(&empty),
&BTreeMap::new(),
LIST_CAP,
);
report.journal = set.journal.as_ref().map(|j| TagJournal {
run_id: j.run_id.clone(),
state: j.state.as_str().to_string(),
started_at: j.started_at.clone(),
});
let restored_sha = set
.project_zip
.as_ref()
.and_then(|zip| live_digest(&loaded.file.project, zip).ok())
.map(|d| d.sha256);
DeployProjectsResult {
env: input.env.clone(),
gateway: loaded.target.gateway.clone(),
project: loaded.file.project.clone(),
release: set.run.release.clone(),
check,
action: action_of(&set.run.project_action),
would_change: true,
artifact_sha256: restored_sha
.or_else(|| live_sha256.clone())
.unwrap_or_default(),
live_sha256,
members: MemberDelta::default(),
removed: Vec::new(),
removed_count: 0,
excluded: Vec::new(),
excluded_count: 0,
unmanaged: Vec::new(),
unmanaged_count: 0,
parent: loaded.meta.parent.clone(),
previous_release: None,
recorded_sha256: None,
live_matches_last_deploy: None,
allow_downgrade: false,
allow_first_change: None,
warnings: Vec::new(),
run_id: None,
run_dir: None,
record_written: false,
restore_command: None,
children: Vec::new(),
missing_tags: None,
missing_tags_count: None,
missing_tags_unknown_providers: None,
missing_tags_unknown_providers_count: None,
missing_tags_unavailable: None,
missing_tags_limit: MISSING_TAGS_LIMIT,
tags: Some(Box::new(report)),
outcome: if check {
ProjectOutcome::Checked
} else {
ProjectOutcome::RolledBack
},
ok: true,
slug: None,
}
}
#[allow(clippy::too_many_arguments)]
async fn rollback_run_preview(
gateway: &dyn ProjectGateway,
loaded: &LoadedProject,
set: &HostSet,
writes: &TagWrites,
groups: Option<&GroupsDone>,
target: &RunId,
route: &RouteState<'_>,
result: &mut DeployProjectsResult,
) {
let project = loaded.file.project.as_str();
let mut reimported = Vec::new();
let mut carried = Vec::new();
let mut undeletable = Vec::new();
let mut deleted = Vec::new();
let mut project_reimported = false;
if let Some(zip) = &set.project_zip
&& let Ok(pre) = live_digest(project, zip)
{
project_reimported = result.live_sha256.as_deref() != Some(pre.sha256.as_str());
}
if set.run.project_action == "create" && result.live_sha256.is_some() {
result.warnings.push(format!(
"project {project} was created by run {} and stays: the gateway surface has no \
project delete",
target.as_str()
));
}
let empty = BTreeMap::new();
let no_record = BTreeSet::new();
for provider in &set.run.providers {
let name = provider.name.as_str();
if writes.created.contains_key(name) {
continue;
}
let (Some(applied), Some(plan)) = (
writes.applied.get(name),
writes
.plan
.as_ref()
.and_then(|p| p.providers.iter().find(|pp| pp.provider == name)),
) else {
continue;
};
let current = match read_flat(gateway, name).await {
Ok(current) => current,
Err(err) => {
result.warnings.push(format!(
"[{name}]: the provider could not be read to plan its undo ({})",
err.code()
));
continue;
}
};
let pre = set.pre.get(name).unwrap_or(&empty);
let recreate = writes.recreate.get(name).unwrap_or(&no_record);
match rollback_plan(
plan,
pre,
¤t,
applied,
false,
recreate,
route.ready().map(|(_, handshake)| handshake),
TAG_DELETE_DESCENDANTS,
) {
Ok(undo) => {
for batch in &undo.imports {
reimported.extend(batch.nodes.iter().map(|p| full_tag_path(name, p)));
}
carried.extend(undo.carried.iter().map(|p| full_tag_path(name, p)));
undeletable.extend(undo.undeletable.iter().map(|p| full_tag_path(name, p)));
for (path, why) in &undo.foreign {
undeletable.push(full_tag_path(name, path));
result.warnings.push(format!(
"{}: {}",
full_tag_path(name, path),
why.describe()
));
}
deleted.extend(undo.deletes.iter().map(|p| full_tag_path(name, p)));
for path in &undo.uncarriable {
result.warnings.push(format!(
"{}: its live value cannot be carried into an Overwrite, so the undo \
would leave it and end rollback_failed",
full_tag_path(name, path)
));
}
}
Err(err) => result.warnings.push(format!(
"[{name}]: its undo could not be planned ({})",
err.code()
)),
}
}
let mut provider_restored = Vec::new();
for (name, pre) in &writes.modified {
if let Ok(Some(now)) = gateway.tag_provider_record(name).await
&& !settings_equal(&now, pre)
{
provider_restored.push(name.clone());
}
}
let mut provider_deleted = Vec::new();
for name in writes.created.keys() {
if let Ok(Some(_)) = gateway.tag_provider_record(name).await {
provider_deleted.push(name.clone());
}
}
if let Some(GroupsDone::Unrestored(why)) = groups {
result.warnings.push(why.clone());
}
if !undeletable.is_empty() {
result.warnings.push(format!(
"the undo would end rollback_failed: {} node(s) run {} created cannot be removed \
(data.tags.delete_route says whether the tag-delete route can be used)",
undeletable.len(),
target.as_str()
));
}
if let Some(back) = &writes.records_back {
for (provider, bytes) in back {
if bytes.is_none() {
result.warnings.push(format!(
"tag provider {provider} had no tag record before run {}: the record that \
run left, if any, would stay",
target.as_str()
));
}
}
}
let (reimported, reimported_count) = sorted_capped(reimported);
let (carried, carried_count) = sorted_capped(carried);
let (undeletable, undeletable_count) = sorted_capped(undeletable);
let (deleted, deleted_count) = sorted_capped(deleted);
provider_restored.sort_by(|a, b| a.as_bytes().cmp(b.as_bytes()));
result.would_change = project_reimported
|| reimported_count > 0
|| deleted_count > 0
|| !provider_restored.is_empty()
|| !provider_deleted.is_empty()
|| matches!(groups, Some(GroupsDone::Committed { .. }));
if let Some(tags) = result.tags.as_mut() {
tags.rollback = Some(TagRollback {
cause: RollbackCause::RollbackRun.as_str().to_string(),
detail: format!(
"what the undo of run {} from the host copy would restore; nothing was written",
target.as_str()
),
reimported,
reimported_count,
carried,
carried_count,
undeletable,
undeletable_count,
groups_restored: matches!(groups, Some(GroupsDone::Committed { .. })),
provider_restored,
provider_deleted,
project_reimported,
verify: None,
deleted,
deleted_count,
});
}
}
pub async fn deploy_projects_rollback_run(
loaded: &LoadedProject,
input: &ProjectsInput,
surfaces: &ProjectsSurfaces<'_>,
opts: &ProjectsRunOptions<'_>,
run_id: &str,
check: bool,
observer: &dyn ProjectsObserver,
) -> Result<DeployProjectsResult, CoreError> {
let target = RunId::parse(run_id)?;
if loaded.owned_tags.is_empty() {
return Err(CoreError::InvalidInput {
reason: "--rollback-run undoes a tag-owning run, and deploy/project.yaml lists no \
tag_providers"
.to_string(),
});
}
let project = loaded.file.project.as_str();
let gateway = surfaces.gateway;
let git = surfaces.host.git_entry(project).await?;
if git.is_present() {
return Err(CoreError::ProjectGitManaged {
project: project.to_string(),
location: git_entry_location(project),
kind: git.as_str().to_string(),
});
}
let set = read_host_set(loaded, input, surfaces.host, &target).await?;
let mut writes = undo_writes(&set);
let project_undo = match set.run.project_action.as_str() {
"import" => Some(ProjectUndo {
pre: set.project_zip.clone(),
}),
"create" => Some(ProjectUndo { pre: None }),
_ => None,
};
let live_sha256 = project_sha(gateway, project).await;
let built_host;
let apply_host: Option<&dyn ApplyHost> = match surfaces.apply_host {
Some(host) => Some(host),
None if set.run.groups => {
built_host = apply_host_for_node(
&opts.targets_root,
&opts.target_file,
"nodes.master",
&opts.master,
&loaded.target.gateway,
&loaded.target.mode,
)?;
Some(built_host.as_ref())
}
None => None,
};
writes.groups = undo_groups(&set, input, apply_host, &target).await;
let mut result = rollback_run_result(loaded, input, &set, &writes, live_sha256, check);
let route = match &surfaces.tag_delete {
TagDeleteSurface::Configured(client) => {
let answer = if check {
client.handshake().await
} else {
wait_ready(*client, opts.tag_delete_ready, opts.tag_delete_poll).await
};
RouteState::of(&surfaces.tag_delete, Some(answer))
}
surface => RouteState::of(surface, None),
};
if let Some(tags) = result.tags.as_mut() {
tags.delete_route = Some(route.report(&owned_names(loaded)));
}
if let (Some(_), None) = (route.route, &route.handshake) {
result.warnings.push(format!(
"the tag-delete route is not ready ({}); the undo cannot remove the tags run {} \
created",
route.problem.as_deref().unwrap_or("no handshake"),
target.as_str()
));
}
if check {
rollback_run_preview(
gateway,
loaded,
&set,
&writes,
writes.groups.as_ref(),
&target,
&route,
&mut result,
)
.await;
return Ok(result);
}
let built_run;
let run_writer: &dyn ProjectRunWriter = match opts.run_writer {
Some(writer) => writer,
None => {
built_run = project_run_writer_for_node(
&opts.targets_root,
&opts.target_file,
"nodes.master",
&opts.master,
)?;
built_run.as_ref()
}
};
let built_tags;
let tag_writer: &dyn TagRecordWriter = match opts.tag_record_writer {
Some(writer) => writer,
None => {
built_tags = tag_record_writer_for_node(
&opts.targets_root,
&opts.target_file,
"nodes.master",
&opts.master,
)?;
built_tags.as_ref()
}
};
let built_release;
let release_writer: &dyn ReleaseRecordWriter = match opts.record_writer {
Some(writer) => writer,
None => {
built_release = release_writer_for_node(
&opts.targets_root,
&opts.target_file,
"nodes.master",
&opts.master,
)?;
built_release.as_ref()
}
};
wrong_target_check(loaded, gateway, opts).await?;
let designers = if project_undo.as_ref().is_some_and(|undo| undo.pre.is_some()) {
designers_check(&result, gateway, &opts.designers, observer).await?
} else {
DesignersFound {
mode: "skipped",
timeout_s: None,
polls: Vec::new(),
seen: Vec::new(),
proceeded: false,
skipped: true,
}
};
let (rid, run_dir) = RunDir::create_fresh(&opts.repo, &loaded.target.gateway)?;
let rel = run_dir_rel(&loaded.target.gateway, &rid);
result.run_id = Some(rid.as_str().to_string());
result.run_dir = Some(rel.clone());
let surfaces = &ProjectsSurfaces {
gateway: surfaces.gateway,
host: surfaces.host,
tree: surfaces.tree,
apply_host,
deploy: surfaces.deploy,
tag_delete: surfaces.tag_delete.clone(),
};
let outcome = async {
run_dir.write_json(
"run.json",
&serde_json::json!({
"kind": "rollback_run",
"run_id": rid.as_str(),
"target_run_id": target.as_str(),
"started_at": rfc3339_now_utc(),
"env": input.env,
"gateway": loaded.target.gateway,
"project": project,
"release": set.run.release,
"tag_providers": loaded.file.tag_providers,
"designers": designers.mode,
"designers_seen": sessions_json(&designers.seen),
"run_dir_git": run_dir.git_status().as_record(),
}),
)?;
let mut events = RunEvents {
log: EventLog::open(&run_dir)?,
observer,
late_failure: false,
};
events.emit(ProjectsEvent::RunStarted {
run_id: rid.as_str().to_string(),
action: result.action,
})?;
if designers.proceeded {
events.emit(ProjectsEvent::DesignersProceed {
sessions: designers.seen.clone(),
})?;
}
let journal = ProjectJournal {
schema_version: 1,
gateway: loaded.target.gateway.clone(),
project: project.to_string(),
run_id: target.as_str().to_string(),
state: JournalState::RollingBack,
started_at: rfc3339_now_utc(),
providers: set.run.providers.iter().map(|p| p.name.clone()).collect(),
};
run_writer
.write_journal(&loaded.target.gateway, project, &journal.to_bytes())
.await?;
events.emit(ProjectsEvent::JournalWritten {
state: JournalState::RollingBack.as_str(),
})?;
writes.host = Some(HostCopy {
run: set.run.clone(),
journal,
});
let failure = TagFailure {
cause: RollbackCause::RollbackRun,
what: format!("the undo of run {} from the host copy", target.as_str()),
writes,
};
roll_back(
failure,
project_undo,
result,
loaded,
input,
surfaces,
opts,
&rid,
&rel,
UndoWriters {
run: Some(run_writer),
records: Some(tag_writer),
release: Some(release_writer),
},
&route,
&mut events,
)
.await
}
.await;
match outcome {
Ok(mut result) => {
if let Err(err) = run_dir.write_json("result.json", &result) {
result.warnings.push(format!(
"the run's result.json could not be written ({})",
err.code()
));
}
Ok(result)
}
Err(err) => {
let _ = run_dir.write_json(
"result.json",
&serde_json::json!({
"ok": false,
"slug": err.code(),
"run_id": rid.as_str(),
"target_run_id": target.as_str(),
}),
);
Err(err)
}
}
}
const GROUP_SCAN_POLL: Duration = Duration::from_millis(500);
const GROUP_SCAN_WAIT: Duration = Duration::from_secs(120);
fn group_scan_poll() -> crate::poll::PollConfig {
crate::poll::PollConfig {
subject: "the config scan after the tag-group commit".to_string(),
interval: GROUP_SCAN_POLL,
max: Duration::from_secs(2),
deadline: GROUP_SCAN_WAIT,
}
}
struct GroupWork {
providers: Vec<String>,
plans: Vec<GroupPlan>,
}
impl GroupWork {
fn would_change(&self) -> bool {
self.plans.iter().any(GroupPlan::would_change)
}
}
async fn fresh_groups(
loaded: &LoadedProject,
input: &ProjectsInput,
surfaces: &ProjectsSurfaces<'_>,
report: &TagsReport,
tag_plan: &TagPlan,
tag_records: &std::collections::BTreeMap<String, TagRecord>,
live: &ProviderReads,
) -> Result<Option<GroupWork>, CoreError> {
if report.groups.is_empty() {
return Ok(None);
}
let reader = need_tree(surfaces, "the tag-group write")?;
let inputs = group_inputs(loaded, tag_plan)?;
let providers: Vec<String> = report.groups.iter().map(|g| g.provider.clone()).collect();
let observed = read_groups(reader, &loaded.target.mode, &providers).await?;
let plans = group_plans(&inputs, &observed, tag_records, live)?;
if plans != report.groups {
let shown = |plans: &[GroupPlan]| {
plans
.iter()
.map(|p| {
format!(
"{}: {} added, {} changed, {} removed",
p.provider,
p.added.len(),
p.changed.len(),
p.removed.len()
)
})
.collect::<Vec<_>>()
.join("; ")
};
return Err(CoreError::PlanStale {
expected: shown(&report.groups),
actual: shown(&plans),
detail: "the owned providers' tag groups changed since the plan".to_string(),
});
}
let work = GroupWork { providers, plans };
if work.would_change() {
let gateway = &loaded.target.gateway;
let mode = &loaded.target.mode;
if let Some(bytes) = reader.read_journal(gateway, mode).await? {
let location = journal_location(
LedgerRoot::for_node(&loaded.target.nodes.master),
gateway,
mode,
);
let journal = Journal::parse(&location, &bytes, gateway, mode)?;
return Err(CoreError::DeployJournalPresent {
run_id: journal.run_id,
env: input.env.clone(),
location,
recover_with: None,
});
}
}
Ok(Some(work))
}
fn group_batch_spec(
loaded: &LoadedProject,
run_id: &RunId,
plan_sha256: &str,
plans: &[GroupPlan],
) -> (BatchSpec, RenderedTree) {
let mut tree = RenderedTree::default();
let mut writes: Vec<WriteOp> = Vec::new();
let mut deletes: Vec<DeleteOp> = Vec::new();
for plan in plans {
for write in &plan.writes {
writes.push(WriteOp {
path: write.path.clone(),
sha256: sha256_hex(&write.bytes),
pre: write.pre.clone(),
});
tree.files.insert(write.path.clone(), write.bytes.clone());
}
for delete in &plan.deletes {
deletes.push(DeleteOp {
path: delete.path.clone(),
pre: delete.pre.clone(),
});
}
}
writes.sort_by(|a, b| a.path.as_str().as_bytes().cmp(b.path.as_str().as_bytes()));
deletes.sort_by(|a, b| a.path.as_str().as_bytes().cmp(b.path.as_str().as_bytes()));
let mut touched: Vec<TouchedEntry> = writes
.iter()
.map(|w| TouchedEntry {
path: w.path.as_str().to_string(),
pre_sha256: w.pre.clone(),
post_sha256: Some(w.sha256.clone()),
})
.chain(deletes.iter().map(|d| TouchedEntry {
path: d.path.as_str().to_string(),
pre_sha256: Some(d.pre.clone()),
post_sha256: None,
}))
.collect();
touched.sort_by(|a, b| a.path.as_bytes().cmp(b.path.as_bytes()));
let pre_fingerprint = fingerprint(
&touched
.iter()
.map(|e| (e.path.clone(), e.pre_sha256.clone()))
.collect(),
);
let gateway = loaded.target.gateway.clone();
let mode = loaded.target.mode.clone();
let run = run_id.as_str().to_string();
let budget = HOLD_S.saturating_sub(LOCK_MARGIN_S.unsigned_abs()).max(1);
let spec = BatchSpec {
run_id: run_id.clone(),
gateway: gateway.clone(),
mode: mode.clone(),
writes,
deletes,
ledger_new: None,
ledger_pre: None,
journal: Journal {
schema_version: 1,
run_id: run.clone(),
gateway: gateway.clone(),
mode: mode.clone(),
phase: JournalPhase::Snapshotting,
pre_fingerprint: pre_fingerprint.clone(),
touched: touched.iter().map(|e| e.path.clone()).collect(),
},
record: RunRecord {
schema_version: 1,
run_id: run,
gateway,
mode,
render_sha256: tree.render_sha256(),
plan_sha256: plan_sha256.to_string(),
state: RecordState::Applying,
rolled_back_by: None,
touched,
pre_fingerprint,
post_fingerprint: None,
},
commit_budget_s: Some(u32::try_from(budget).unwrap_or(u32::MAX)),
};
(spec, tree)
}
enum GroupFailure {
NotWritten(CoreError),
Stale,
Restored(String),
Unrestored(String),
}
async fn restore_groups(
host: &dyn ApplyHost,
deploy: &dyn DeployApi,
record: &RunRecord,
run_id: &RunId,
lock: Option<&HeldLock>,
) -> Result<(), String> {
let held = match lock {
Some(lock) => matches!(
lock.check(deploy, LOCK_MARGIN_S, &SystemClock).await,
Ok(LockVerdict::Held { .. })
),
None => false,
};
if !held {
HeldLock::acquire(deploy, ScanKind::Config, HOLD_S, &SystemClock)
.await
.map_err(|e| format!("taking the scan-lock for the restore: {}", e.code()))?;
}
let report = host
.restore(record)
.await
.map_err(|e| format!("restore: {}", e.code()))?;
if let Some(failure) = report.failure {
return Err(format!("restore: {failure}"));
}
match scan_and_wait(deploy, ScanKind::Config, group_scan_poll()).await {
Ok(ScanOutcome::Complete(_)) => {}
Ok(ScanOutcome::NotComplete { .. }) => {
return Err("the scan after the restore did not complete".to_string());
}
Err(e) => return Err(format!("the scan after the restore: {}", e.code())),
}
let mut rolled_back = record.clone();
rolled_back.state = RecordState::RolledBack;
rolled_back.rolled_back_by = Some(run_id.as_str().to_string());
rolled_back.post_fingerprint = Some(record.pre_fingerprint.clone());
host.finalize(&rolled_back)
.await
.map_err(|e| format!("finalize: {}", e.code()))
}
async fn restored(
host: &dyn ApplyHost,
deploy: &dyn DeployApi,
record: &RunRecord,
run_id: &RunId,
lock: &HeldLock,
why: String,
) -> GroupFailure {
match restore_groups(host, deploy, record, run_id, Some(lock)).await {
Ok(()) => GroupFailure::Restored(why),
Err(cause) => GroupFailure::Unrestored(format!("{why}; {cause}")),
}
}
async fn commit_groups(
loaded: &LoadedProject,
surfaces: &ProjectsSurfaces<'_>,
run_id: &RunId,
plan_sha256: &str,
work: &GroupWork,
) -> Result<(usize, usize, RunRecord), GroupFailure> {
let missing = |what: &str| {
GroupFailure::NotWritten(CoreError::Internal(format!(
"the tag-group write needs the master's {what}, and none was supplied"
)))
};
let host = surfaces.apply_host.ok_or_else(|| missing("apply host"))?;
let deploy = surfaces.deploy.ok_or_else(|| missing("deploy API"))?;
let reader = surfaces.tree.ok_or_else(|| missing("host reader"))?;
let (spec, tree) = group_batch_spec(loaded, run_id, plan_sha256, &work.plans);
let touched = spec.touched_paths();
let probe = host
.probe_paths(&touched)
.await
.map_err(GroupFailure::NotWritten)?;
if !probe.writable {
return Err(GroupFailure::NotWritten(CoreError::TransportFailed {
operation: "write".to_string(),
node: "nodes.master".to_string(),
detail: format!(
"the {} collection cannot take the tag-group units ({})",
spec.mode,
probe.problems.join("; ")
),
}));
}
let prepared = host
.prepare(&spec, &tree)
.await
.map_err(GroupFailure::NotWritten)?;
let lock = match HeldLock::acquire(deploy, ScanKind::Config, HOLD_S, &SystemClock).await {
Ok(lock) => lock,
Err(e) => {
let _ = host.discard_interrupted(run_id).await;
return Err(GroupFailure::NotWritten(e));
}
};
let report = host
.commit(&prepared)
.await
.map_err(GroupFailure::NotWritten)?;
if report.stale {
let _ = scan_and_wait(deploy, ScanKind::Config, group_scan_poll()).await;
return Err(GroupFailure::Stale);
}
let restore = |why: String| restored(host, deploy, &spec.record, run_id, &lock, why);
if let Some(failure) = &report.failure {
return Err(restore(format!("the commit stopped: {failure}")).await);
}
match lock.check(deploy, LOCK_MARGIN_S, &SystemClock).await {
Ok(LockVerdict::Held { .. }) => {}
Ok(_) => return Err(restore("the scan-lock lapsed during the commit".to_string()).await),
Err(e) => {
return Err(restore(format!("the scan-lock could not be read ({})", e.code())).await);
}
}
match scan_and_wait(deploy, ScanKind::Config, group_scan_poll()).await {
Ok(ScanOutcome::Complete(_)) => {}
Ok(ScanOutcome::NotComplete { .. }) => {
return Err(restore("the config scan did not complete".to_string()).await);
}
Err(e) => return Err(restore(format!("the config scan failed ({})", e.code())).await),
}
let observed = match reader
.read_tag_groups(&loaded.target.mode, &work.providers)
.await
{
Ok(observed) => observed,
Err(e) => {
return Err(restore(format!("the units could not be read back ({})", e.code())).await);
}
};
let mut differing: Vec<String> = spec
.writes
.iter()
.filter(|w| observed.mode.get(&w.path).map(|b| b.sha256.as_str()) != Some(&w.sha256))
.map(|w| w.path.as_str().to_string())
.collect();
differing.extend(
spec.deletes
.iter()
.filter(|d| observed.mode.contains_key(&d.path))
.map(|d| d.path.as_str().to_string()),
);
if !differing.is_empty() {
return Err(restore(format!(
"{} unit file(s) read back different: {}",
differing.len(),
differing.join(", ")
))
.await);
}
let post = match host.fingerprint(&touched).await {
Ok(post) => post,
Err(e) => {
return Err(GroupFailure::Unrestored(format!(
"the units verified but their hashes could not be read ({}); the journal stays",
e.code()
)));
}
};
let mut record = spec.record.clone();
record.state = RecordState::Applied;
record.post_fingerprint = Some(fingerprint(&post));
host.finalize(&record).await.map_err(|e| {
GroupFailure::Unrestored(format!(
"the units verified but the run record could not be finalized ({}); the journal stays",
e.code()
))
})?;
Ok((spec.writes.len(), spec.deletes.len(), record))
}
async fn fresh_provider_settings(
loaded: &LoadedProject,
gateway: &dyn ProjectGateway,
report: &TagsReport,
run_dir: &RunDir,
run_id: &RunId,
) -> Result<Vec<(ProviderSettingsPlan, Option<TagProviderRecord>)>, CoreError> {
let mut out = Vec::new();
for planned in &report.provider_settings {
let name = planned.provider.as_str();
let record = gateway.tag_provider_record(name).await?;
let now =
plan_provider_settings(name, provider_settings_of(loaded, name)?, record.as_ref());
if &now != planned {
let shown = |p: &ProviderSettingsPlan| {
if p.properties.is_empty() {
p.action.as_str().to_string()
} else {
format!("{} ({})", p.action.as_str(), p.properties.join(", "))
}
};
return Err(CoreError::PlanStale {
expected: shown(planned),
actual: shown(&now),
detail: format!("the settings of tag provider {name} changed since the plan"),
});
}
if planned.action == ProviderSettingsAction::Modify
&& let Some(record) = &record
{
let bytes = serde_json::to_vec_pretty(record).map_err(|e| {
CoreError::Internal(format!("a tag provider record did not serialize: {e}"))
})?;
let mut file = run_dir.create_private_file_nested(
&["export", "tags"],
&format!("{name}.provider.json"),
)?;
file.write_all(&bytes)
.and_then(|()| file.sync_all())
.map_err(|e| CoreError::DeployRunInvalid {
run_id: run_id.as_str().to_string(),
problem: format!(
"could not write the settings pre-image of tag provider {name} ({:?})",
e.kind()
),
})?;
}
out.push((planned.clone(), record));
}
Ok(out)
}
fn note_tag_step(result: &mut DeployProjectsResult, step: &TagStep) {
let Some(tags) = result.tags.as_mut() else {
return;
};
tags.carried_count = step.carried.len();
tags.carried = step.carried.iter().take(LIST_CAP).cloned().collect();
tags.verify = step.verify.clone();
tags.deleted_count = step.deleted.len();
tags.deleted = step.deleted.iter().take(LIST_CAP).cloned().collect();
if !step.absent.is_empty() {
tags.already_absent_count += step.absent.len();
tags.already_absent.extend(step.absent.iter().cloned());
tags.already_absent
.sort_by(|a, b| a.as_bytes().cmp(b.as_bytes()));
tags.already_absent.dedup();
tags.already_absent.truncate(LIST_CAP);
}
if step.verify.as_ref().is_some_and(|v| !v.settled) {
result.warnings.push(format!(
"the owned tag providers did not read the same twice within the verify bound; \
the last read was compared and equals the repo ({} read(s))",
step.verify.as_ref().map_or(0, |v| v.reads)
));
}
}
async fn write_tag_records(
result: &mut DeployProjectsResult,
loaded: &LoadedProject,
run_id: &RunId,
writer: Option<&dyn TagRecordWriter>,
events: &mut RunEvents<'_>,
changed_gateway: bool,
) -> Result<(), CoreError> {
let (Some(tags), Some(writer)) = (result.tags.as_ref(), writer) else {
return Ok(());
};
let tags_sha256 = tags.tags_sha256.clone();
for provider in sorted(&loaded.owned_tags) {
let mut groups: Vec<String> = provider
.groups
.as_ref()
.and_then(|g| g.get("tagGroups"))
.and_then(Value::as_array)
.into_iter()
.flatten()
.filter_map(|g| g.get("name").and_then(Value::as_str))
.map(str::to_string)
.collect();
groups.sort_by(|a, b| a.as_bytes().cmp(b.as_bytes()));
groups.dedup();
let record = TagRecord {
schema_version: 1,
gateway: result.gateway.clone(),
provider: provider.name.clone(),
project: result.project.clone(),
release: loaded.release.tag.clone(),
run_id: run_id.as_str().to_string(),
deployed_at: rfc3339_now_utc(),
tags_sha256: tags_sha256.clone(),
nodes: provider.nodes.keys().cloned().collect(),
groups,
};
match writer
.write_tag_record(&result.gateway, &provider.name, &record.to_bytes())
.await
{
Ok(()) => events.emit_late(ProjectsEvent::TagRecordWritten {
provider: provider.name.clone(),
}),
Err(err) => {
events.emit_late(ProjectsEvent::TagRecordFailed {
provider: provider.name.clone(),
code: err.code().to_string(),
});
if !changed_gateway {
return Err(err);
}
result.warnings.push(format!(
"the run is verified but the tag record of provider {} could not be written ({})",
provider.name,
err.code()
));
}
}
}
Ok(())
}
fn sorted(providers: &[NativeProvider]) -> Vec<&NativeProvider> {
let mut out: Vec<&NativeProvider> = providers.iter().collect();
out.sort_by(|a, b| a.name.as_bytes().cmp(b.name.as_bytes()));
out
}
fn import_outcome_unknown(err: &CoreError) -> bool {
match err {
CoreError::Network {
source: Some(source),
..
} => !(source.is_connect() || source.is_builder()),
CoreError::Network { source: None, .. } | CoreError::Internal(_) => true,
_ => false,
}
}
fn finish_recorded(
result: &mut DeployProjectsResult,
events: &mut RunEvents<'_>,
backup_unsynced: bool,
record_written: bool,
) {
result.record_written = record_written;
result.outcome = if result.tags.is_some() {
ProjectOutcome::Applied
} else {
ProjectOutcome::Recorded
};
if backup_unsynced && result.tags.is_some() {
mark_backup_unsynced(result);
}
if record_written {
events.emit_late(ProjectsEvent::RecordWritten);
}
events.emit_late(ProjectsEvent::Finished {
outcome: result.outcome,
});
note_late_log_failure(result, events);
}
fn mark_backup_unsynced(result: &mut DeployProjectsResult) {
result.outcome = ProjectOutcome::AppliedBackupUnsynced;
result.ok = false;
result.slug = Some(BACKUP_UNSYNCED_SLUG.to_string());
}
fn backup_event(event: TagsBackupEvent) -> ProjectsEvent {
match event {
TagsBackupEvent::WaitStarted { budget_s } => ProjectsEvent::BackupWaitStarted { budget_s },
TagsBackupEvent::Resync { ok, code } => ProjectsEvent::BackupResync { ok, code },
TagsBackupEvent::Verdict { converged, reason } => ProjectsEvent::BackupVerdict {
converged,
reason: reason.map(|reason| reason.as_str()),
},
}
}
async fn backup_step(
result: &mut DeployProjectsResult,
providers: &[&str],
surfaces: &ProjectsSurfaces<'_>,
opts: &ProjectsRunOptions<'_>,
since_ms: i64,
events: &mut RunEvents<'_>,
) -> Option<bool> {
let backup = opts.backup?;
result.tags.as_ref()?;
let mut master = BTreeMap::new();
let mut unread = None;
for name in providers {
match surfaces.gateway.tag_export(name).await {
Ok(bytes) => match serde_json::from_slice::<Value>(&bytes) {
Ok(tree) => {
master.insert((*name).to_string(), tree);
}
Err(_) => {
unread = Some(format!("[{name}] is not JSON"));
break;
}
},
Err(err) => {
unread = Some(format!("[{name}] {}", err.code()));
break;
}
}
}
let report = match unread {
None => {
let resync = surfaces.deploy.map(DeployResync);
converge_tags(
&master,
backup,
resync.as_ref().map(|r| r as &dyn MasterResync),
opts.backup_policy,
since_ms,
&mut |event| events.emit_late(backup_event(event)),
)
.await
}
Some(why) => {
result.warnings.push(format!(
"the backup check could not read the master's export ({why}), so the backup's \
copy of the owned tags is not proven"
));
let mut names: Vec<String> = providers.iter().map(|p| (*p).to_string()).collect();
names.sort_by(|a, b| a.as_bytes().cmp(b.as_bytes()));
TagsBackupReport {
converged: false,
reason: Some(crate::deploy::pair::UnsyncedReason::Unreachable),
budget_s: opts.backup_policy.budget.as_secs(),
elapsed_s: 0,
resync_sent: false,
providers: names,
differing: Vec::new(),
}
}
};
let converged = report.converged;
if let Some(tags) = result.tags.as_mut() {
tags.backup = Some(report);
}
Some(converged)
}
fn note_late_log_failure(result: &mut DeployProjectsResult, events: &RunEvents<'_>) {
if events.late_failure {
result
.warnings
.push("the run's events log could not be written to the end".to_string());
}
}
#[cfg(test)]
pub(crate) fn sample_result() -> DeployProjectsResult {
DeployProjectsResult {
env: "prod".to_string(),
gateway: "gw".to_string(),
project: "CapApp".to_string(),
release: "v1.0.0".to_string(),
check: false,
action: ProjectAction::Record,
would_change: true,
artifact_sha256: "a".repeat(64),
live_sha256: Some("a".repeat(64)),
members: MemberDelta::default(),
removed: Vec::new(),
removed_count: 0,
excluded: Vec::new(),
excluded_count: 0,
unmanaged: Vec::new(),
unmanaged_count: 0,
parent: None,
previous_release: None,
recorded_sha256: None,
live_matches_last_deploy: None,
allow_downgrade: false,
allow_first_change: None,
warnings: Vec::new(),
run_id: None,
run_dir: None,
record_written: false,
restore_command: None,
children: Vec::new(),
missing_tags: Some(Vec::new()),
missing_tags_count: Some(0),
missing_tags_unknown_providers: Some(Vec::new()),
missing_tags_unknown_providers_count: Some(0),
missing_tags_unavailable: None,
missing_tags_limit: MISSING_TAGS_LIMIT,
tags: None,
outcome: ProjectOutcome::Checked,
ok: true,
slug: None,
}
}
#[cfg(test)]
mod tests {
use super::*;
fn sample_zip() -> Vec<u8> {
use std::io::Write as _;
let mut writer = zip::ZipWriter::new(std::io::Cursor::new(Vec::new()));
let options = zip::write::SimpleFileOptions::default()
.compression_method(zip::CompressionMethod::Deflated);
for (name, bytes) in [
("project.json", br#"{"title":"T"}"#.as_slice()),
("ignition/s/resource.json", br#"{"files":["a.py"]}"#),
("ignition/s/a.py", &[b'x'; 4096]),
] {
writer.start_file(name, options).expect("start");
writer.write_all(bytes).expect("write");
}
writer.finish().expect("finish").into_inner()
}
#[test]
fn check_artifact_zip_refuses_truncated() {
let whole = sample_zip();
check_artifact_zip(&whole).expect("the whole zip passes");
let half = &whole[..whole.len() / 2];
let err = check_artifact_zip(half).expect_err("half a zip");
assert_eq!(err.code(), "invalid_import_file");
let err = check_artifact_zip(&whole[..whole.len() - 40]).expect_err("tail cut");
assert_eq!(err.code(), "invalid_import_file");
let err = check_artifact_zip(b"not a zip").expect_err("no magic");
assert_eq!(err.code(), "invalid_import_file");
}
use super::sample_result;
#[test]
fn a_log_failure_after_the_record_is_written_is_a_warning() {
let dir = tempfile::tempdir().expect("tempdir");
let path = dir.path().join("events.ndjson");
std::fs::write(&path, b"").expect("seed");
let read_only = std::fs::File::open(&path).expect("open read-only");
let mut events = RunEvents {
log: EventLog::from_file(read_only),
observer: &NoProjectsObserver,
late_failure: false,
};
let mut result = sample_result();
finish_recorded(&mut result, &mut events, false, true);
assert!(result.record_written, "the record is on the host");
assert_eq!(result.outcome, ProjectOutcome::Recorded);
assert!(result.ok, "a log failure does not fail the run");
assert!(events.late_failure);
assert!(
result
.warnings
.iter()
.any(|w| w.contains("events log could not be written")),
"{:?}",
result.warnings
);
}
#[test]
fn a_built_artifact_passes_the_check() {
let dir = tempfile::tempdir().expect("tempdir");
for (rel, bytes) in [
("project.json", br#"{"title":"T"}"#.as_slice()),
("ignition/s/resource.json", br#"{"files":["a.py"]}"#),
("ignition/s/a.py", b"pass"),
] {
let path = dir.path().join(rel);
std::fs::create_dir_all(path.parent().expect("parent")).expect("mkdir");
std::fs::write(path, bytes).expect("write");
}
let artifact = build_artifact(dir.path()).expect("artifact");
check_artifact_zip(&artifact.zip).expect("a built artifact is a valid import");
}
}