use std::collections::BTreeSet;
use aion::{AdmissionReason, QueueAdmission, ReasonCensus};
use super::{Disagreement, WorkerAdvertisement};
pub(super) fn render_mismatch(
admission: &QueueAdmission,
node: Option<&str>,
advertised: WorkerAdvertisement<'_>,
disagreements: &[Disagreement],
) -> String {
let mut sections = vec![
checked_summary(admission),
advertisement_summary(advertised),
];
if let Some(node) = node {
sections.push(format!(
"this connection advertises node `{node}`, so it is held only to the actions a \
dispatch on that node can reach"
));
} else {
sections.push(
"this connection advertises no node, so it is held only to unpinned actions".to_owned(),
);
}
sections.push(pluralized(disagreements.len(), "disagreement"));
let total = disagreements.len();
for (index, disagreement) in disagreements.iter().enumerate() {
sections.push(render_disagreement(index + 1, total, disagreement));
}
sections.push(remedy(admission, advertised, disagreements));
sections.join(". ")
}
fn advertisement_summary(advertised: WorkerAdvertisement<'_>) -> String {
let mut clauses = vec![format!(
"worker advertised {} and {}, and admission compares CONTRACTS",
pluralized(advertised.activity_types.len(), "activity-type name"),
pluralized(advertised.contracts.len(), "typed contract"),
)];
let gap = advertised.names_without_contracts();
if !gap.is_empty() {
clauses.push(format!(
"{} advertised by name with NO contract: {} — a name makes a worker \
selectable for dispatch, a contract makes it admissible",
pluralized(gap.len(), "action"),
backticked(&gap),
));
}
clauses.join("; ")
}
fn checked_summary(admission: &QueueAdmission) -> String {
let census = admission.reason_census();
let mut clauses = vec![format!(
"checked {} ({})",
pluralized(
admission.required.len(),
"reachable deployed package version"
),
census_clause(census)
)];
if !admission.unreachable.is_empty() {
let ignored = admission
.unreachable
.iter()
.map(|contract| {
format!(
"`{}` ({})",
contract.package_version,
workflow_clause(&contract.workflow_types)
)
})
.collect::<Vec<_>>()
.join(", ");
clauses.push(format!(
"ignored {} that nothing can dispatch under: {ignored}",
pluralized(admission.unreachable.len(), "unreachable version")
));
}
clauses.join("; ")
}
fn census_clause(census: ReasonCensus) -> String {
let mut clauses = Vec::new();
if census.route_active > 0 {
clauses.push(format!("{} routes new starts", census.route_active));
}
if census.live_workflow > 0 {
clauses.push(format!("{} has a live workflow run", census.live_workflow));
}
if census.start_in_flight > 0 {
clauses.push(format!(
"{} has a workflow start in flight",
census.start_in_flight
));
}
if clauses.is_empty() {
return "none reachable".to_owned();
}
clauses.join(", ")
}
fn render_disagreement(index: usize, total: usize, disagreement: &Disagreement) -> String {
format!(
"[{index}/{total}] package `{}` ({}, held because {}) action `{}` field `{}` expected {} but worker advertised {}",
disagreement.diff.package_version,
workflow_clause(&disagreement.workflow_types),
disagreement.reason.explanation(),
disagreement.diff.action,
disagreement.diff.field,
rendered_value(disagreement.diff.expected.as_ref()),
rendered_value(disagreement.diff.advertised.as_ref()),
)
}
fn remedy(
admission: &QueueAdmission,
advertised: WorkerAdvertisement<'_>,
disagreements: &[Disagreement],
) -> String {
let mut clauses = vec![
"REMEDY: rebuild and redeploy this worker from the same source as the deployed \
package(s) named above, so it advertises their action shapes"
.to_owned(),
];
let gap = advertised.names_without_contracts();
if !gap.is_empty() {
clauses.push(format!(
"the {} listed above with no contract must announce an input and output schema \
at registration, not just a name — the shapes are the ones the deployed \
package declares",
pluralized(gap.len(), "action")
));
}
let reasons = disagreements
.iter()
.map(|disagreement| disagreement.reason)
.collect::<BTreeSet<_>>();
if reasons.contains(&AdmissionReason::RouteActive) {
clauses.push(
"a version that routes new starts cannot be unloaded — supersede it by deploying a \
newer version of that workflow, or re-point the route at another loaded version with \
POST /deploy/route"
.to_owned(),
);
}
if reasons.contains(&AdmissionReason::LiveWorkflow)
|| reasons.contains(&AdmissionReason::StartInFlight)
{
clauses.push(
"a version a live run is pinned to cannot be unloaded either — it stops binding \
workers only once that run reaches a terminal state"
.to_owned(),
);
}
if !admission.unreachable.is_empty() {
let bodies = admission
.unreachable
.iter()
.map(|contract| {
unload_body(
&contract.package_version.to_string(),
&contract.workflow_types,
)
})
.collect::<Vec<_>>()
.join(" ");
clauses.push(format!(
"the ignored version(s) can be cleared with POST /deploy/unload {bodies} — that is \
hygiene only and will NOT change this refusal"
));
}
clauses.join("; ")
}
fn unload_body(version: &str, workflow_types: &[String]) -> String {
let workflow_type = workflow_types
.first()
.map_or("<workflow_type>", String::as_str);
format!(r#"{{"workflow_type":"{workflow_type}","content_hash":"{version}"}}"#)
}
fn workflow_clause(workflow_types: &[String]) -> String {
match workflow_types {
[] => "no workflow type".to_owned(),
[single] => format!("workflow `{single}`"),
many => format!(
"workflows {}",
many.iter()
.map(|name| format!("`{name}`"))
.collect::<Vec<_>>()
.join(", ")
),
}
}
fn backticked(names: &[String]) -> String {
names
.iter()
.map(|name| format!("`{name}`"))
.collect::<Vec<_>>()
.join(", ")
}
fn pluralized(count: usize, noun: &str) -> String {
if count == 1 {
format!("{count} {noun}")
} else {
format!("{count} {noun}s")
}
}
fn rendered_value(value: Option<&serde_json::Value>) -> String {
value.map_or_else(|| "<missing>".to_owned(), serde_json::Value::to_string)
}