From d5af268096f49c320f4e2ee5977c503078c63cf9 Mon Sep 17 00:00:00 2001 From: Codex Date: Wed, 9 Sep 2026 02:44:52 +0000 Subject: [PATCH 1/3] Deploy immutable images from selected build Machines --- crates/ployz-build/src/image_contexts.rs | 83 ++++ crates/ployz-build/src/lib.rs | 28 ++ crates/ployz-build/src/received_recipe.rs | 29 +- crates/ployz-build/src/remote.rs | 34 ++ crates/ployz-build/src/tests.rs | 3 + crates/ployz-build/src/upload.rs | 1 + crates/ployz/src/cli/mod.rs | 22 +- crates/ployz/src/compose/build.rs | 94 ++++- crates/ployz/src/compose/build_inputs.rs | 36 ++ crates/ployz/src/compose/capture.rs | 14 + crates/ployz/src/compose/mod.rs | 4 +- crates/ployz/src/compose/remote_build.rs | 18 +- crates/ployz/src/compose/remote_steps.rs | 180 +++++++++ crates/ployz/src/deploy/apply.rs | 4 +- crates/ployz/src/deploy/pipeline.rs | 64 +++- crates/ployz/src/handlers/build.rs | 61 +-- crates/ployz/src/handlers/deploy.rs | 61 ++- .../ployz/src/handlers/deploy_remote_tests.rs | 354 ++++++++++++++++++ crates/ployz/src/handlers/deploy_tests.rs | 14 +- crates/ployz/src/image.rs | 11 + crates/ployz/src/image/built.rs | 123 ++++++ crates/ployz/src/lib.rs | 3 + crates/ployz/tests/build.rs | 53 ++- crates/ployz/tests/build_layer3/remote.rs | 150 ++++++++ crates/ployz/tests/deploy_client/support.rs | 73 +++- crates/ployz/tests/support/build.rs | 152 ++++++++ crates/ployzd/src/build.rs | 22 +- crates/ployzd/src/build/tests.rs | 9 +- crates/ployzd/src/docker/mod.rs | 5 +- crates/ployzd/src/docker/peer_pull.rs | 106 ++++-- 30 files changed, 1662 insertions(+), 149 deletions(-) create mode 100644 crates/ployz-build/src/image_contexts.rs create mode 100644 crates/ployz/src/compose/remote_steps.rs create mode 100644 crates/ployz/src/handlers/deploy_remote_tests.rs create mode 100644 crates/ployz/src/image/built.rs create mode 100644 crates/ployz/tests/support/build.rs diff --git a/crates/ployz-build/src/image_contexts.rs b/crates/ployz-build/src/image_contexts.rs new file mode 100644 index 0000000000..861e244695 --- /dev/null +++ b/crates/ployz-build/src/image_contexts.rs @@ -0,0 +1,83 @@ +//! Bind completed Service outputs to BuildKit named contexts. The existing +//! Machine image server supplies bytes directly; no external registry fallback. + +use crate::{BuildError, Request}; +use serde_norway::Value; +use std::{fs, net::IpAddr}; + +pub(crate) fn prepare(request: &Request<'_>) -> Result<(), BuildError> { + if request.image_contexts.is_empty() { + return Ok(()); + } + let invalid = |message: &str| BuildError::Request(message.into()); + let mut document: Value = serde_norway::from_slice( + &fs::read(request.working_dir.join(request.compose_file)).map_err(input_error)?, + ) + .map_err(|_| invalid("invalid captured Build recipe"))?; + let services = document + .get_mut("services") + .and_then(Value::as_mapping_mut) + .ok_or_else(|| invalid("Build recipe has no services"))?; + for target in request.targets { + let build = services + .get_mut(Value::String(target.name.clone())) + .and_then(|service| service.get_mut("build")) + .and_then(Value::as_mapping_mut) + .ok_or_else(|| invalid("Build target has no recipe"))?; + let contexts = build.get("additional_contexts"); + let entries = match contexts { + Some(Value::Mapping(values)) => values + .iter() + .map(|(name, value)| name.as_str().zip(value.as_str())) + .collect::>>(), + Some(Value::Sequence(values)) => values + .iter() + .map(|value| value.as_str().and_then(|value| value.split_once('='))) + .collect::>>(), + None | Some(Value::Null) => Some(Vec::new()), + Some(_) => None, + } + .ok_or_else(|| invalid("invalid additional contexts"))?; + let mut bound = serde_norway::Mapping::new(); + for (name, value) in entries { + if let Some(dependency) = value.strip_prefix("service:") { + let image = request.image_contexts.get(dependency).ok_or_else(|| { + invalid("Service image context has no completed Build output") + })?; + bound.insert( + name.into(), + format!( + "docker-image://{}/{}", + std::net::SocketAddr::new( + image + .source + .management_address + .0 + .to_ipv4_mapped() + .map_or(IpAddr::V6(image.source.management_address.0), IpAddr::V4), + image.source.port, + ), + image.reference + ) + .into(), + ); + } else { + bound.insert(name.into(), value.into()); + } + } + if !bound.is_empty() { + build.insert("additional_contexts".into(), Value::Mapping(bound)); + } + } + fs::write( + request.working_dir.join(request.compose_file), + serde_norway::to_string(&document) + .map_err(|error| BuildError::Request(error.to_string()))?, + ) + .map_err(input_error)?; + Ok(()) +} + +fn input_error(error: std::io::Error) -> BuildError { + BuildError::Request(format!("prepare completed image contexts: {error}")) +} diff --git a/crates/ployz-build/src/lib.rs b/crates/ployz-build/src/lib.rs index a1b5599013..53575a8ba7 100644 --- a/crates/ployz-build/src/lib.rs +++ b/crates/ployz-build/src/lib.rs @@ -12,6 +12,7 @@ mod builder; mod execution; +mod image_contexts; mod received_recipe; pub mod remote; mod upload; @@ -58,6 +59,8 @@ pub fn builder_name() -> String { /// Paths point into the caller's private capture; nothing here is read from /// the original sources again. pub struct Request<'a> { + /// Completed Service image contexts served from their actual Build hosts. + pub image_contexts: &'a BTreeMap, /// Captured Compose file describing every target of this Build. pub compose_file: &'a Path, /// Private directory the build runs in, so no stray file can join it. @@ -116,6 +119,18 @@ pub struct BuiltImage { pub platform: String, } +/// An immutable named image context, delivered by the existing Machine image +/// server. It carries no Machine selection or Deploy policy into the host. +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +pub struct ImageContext { + /// Repository reference pinned to the completed manifest’s SHA-256 digest. + pub reference: String, + /// The single platform actually available in that completed image. + pub platform: String, + /// Open serving endpoint of the Machine containing that exact content. + pub source: ployz_core::ImageIngestDestination, +} + /// Why a Build Attempt did not produce the image it was asked for. #[derive(Clone, Debug, Eq, Error, PartialEq)] pub enum BuildError { @@ -250,12 +265,25 @@ pub fn execute_admitted( docker .run("check Buildx", &["buildx", "version"], Streams::Captured) .map_err(|error| error.at(Stage::Preparation))?; + image_contexts::prepare(request).map_err(|error| error.at(Stage::Preparation))?; let builder = Builder::acquire(&docker, admission.lock).map_err(|error| error.at(Stage::Preparation))?; let result = (|| { let native = builder .native_platform(request.targets) .map_err(|error| error.at(Stage::Preparation))?; + for image in request.image_contexts.values() { + for target in request.targets { + let platform = target.platform.as_deref().unwrap_or(&native); + if !covers(&image.platform, platform) { + return Err(BuildError::Request(format!( + "image context {} contains {}, but this Build requires {platform}", + image.reference, image.platform + )) + .at(Stage::Preparation)); + } + } + } let preparation = railpack::prepare(&docker, request).map_err(|error| error.at(Stage::Preparation))?; let overrides = preparation diff --git a/crates/ployz-build/src/received_recipe.rs b/crates/ployz-build/src/received_recipe.rs index a8b466503e..ab88508905 100644 --- a/crates/ployz-build/src/received_recipe.rs +++ b/crates/ployz-build/src/received_recipe.rs @@ -35,6 +35,28 @@ pub fn validate_capture(root: &Path, definition: &Definition) -> Result<(), Inpu { return Err("invalid Build target names".into()); } + if definition.image_contexts.len() > 128 + || definition.image_contexts.keys().any(|name| { + names.contains(name.as_str()) + || name.is_empty() + || !name + .bytes() + .all(|c| c.is_ascii_alphanumeric() || b"_.-".contains(&c)) + }) + { + return Err("invalid completed Build context names".into()); + } + for context in definition.image_contexts.values() { + validate_remote_context(&format!("docker-image://{}", context.reference))?; + if context.source.port == 0 || !crate::remote::linux_platform(&context.platform) { + return Err("invalid completed Build image context".into()); + } + } + let context_names = names + .iter() + .copied() + .chain(definition.image_contexts.keys().map(String::as_str)) + .collect(); let bytes = fs::read(root.join("compose.yaml")).map_err(|_| "Build recipe is missing")?; let document: Value = serde_norway::from_slice(&bytes).map_err(|_| "invalid captured Build recipe")?; @@ -97,7 +119,7 @@ pub fn validate_capture(root: &Path, definition: &Definition) -> Result<(), Inpu .get("context") .ok_or("Build recipe has no captured context")?, )?; - context_path(root, context, &names, default_ssh)?; + context_path(root, context, &context_names, default_ssh)?; if let Some(recipe) = build.get("dockerfile") { let recipe = text(recipe)?; if remote(context) { @@ -115,7 +137,7 @@ pub fn validate_capture(root: &Path, definition: &Definition) -> Result<(), Inpu match contexts { Value::Mapping(contexts) => { for value in contexts.values() { - context_path(root, text(value)?, &names, default_ssh)?; + context_path(root, text(value)?, &context_names, default_ssh)?; } } Value::Sequence(contexts) => { @@ -123,7 +145,7 @@ pub fn validate_capture(root: &Path, definition: &Definition) -> Result<(), Inpu let (_, value) = text(value)? .split_once('=') .ok_or("invalid named Build context")?; - context_path(root, value, &names, default_ssh)?; + context_path(root, value, &context_names, default_ssh)?; } } Value::Null => {} @@ -397,6 +419,7 @@ mod tests { fs::write(root.join("private/key"), "captured-key").unwrap(); fs::write(root.join("source/key"), "source-content").unwrap(); let definition = Definition { + image_contexts: Default::default(), targets: vec![crate::Target { name: "api".into(), platform: None, diff --git a/crates/ployz-build/src/remote.rs b/crates/ployz-build/src/remote.rs index 17c4bb3d86..495df80e43 100644 --- a/crates/ployz-build/src/remote.rs +++ b/crates/ployz-build/src/remote.rs @@ -8,6 +8,22 @@ use serde::{Deserialize, Serialize, de::DeserializeOwned}; pub use crate::received_recipe::{validate_capture, validate_remote_context}; pub use crate::upload::{Upload, upload}; +/// Whether a completed image reports a well-formed Linux platform. Worker +/// capability checks, rather than a fixed architecture list, determine support. +#[must_use] +pub fn linux_platform(platform: &str) -> bool { + let valid = |part: &str| { + !part.is_empty() + && part + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || b"._-".contains(&byte)) + }; + let mut parts = platform.split('/'); + parts.next() == Some("linux") + && parts.next().is_some_and(valid) + && parts.next().is_none_or(valid) + && parts.next().is_none() +} /// Invalid or unavailable captured input at the transport trust boundary. #[derive(Debug, thiserror::Error)] #[error("{0}")] @@ -32,6 +48,9 @@ pub const CHUNK_SIZE: usize = 32 * 1024; #[derive(Clone, Debug, Serialize, Deserialize)] #[serde(deny_unknown_fields)] pub struct Definition { + /// Completed Service contexts, bound to immutable content and serving endpoints. + #[serde(default)] + pub image_contexts: std::collections::BTreeMap, /// Targets in this admitted attempt. pub targets: Vec, /// Requested disposition of completed output. @@ -135,6 +154,21 @@ impl Outcome { #[cfg(test)] mod tests { + #[test] + fn linux_platform_validation_keeps_worker_architectures_and_rejects_malformed_values() { + for value in ["linux/amd64", "linux/386", "linux/arm/v7", "linux/ppc64le"] { + assert!(super::linux_platform(value), "{value}"); + } + for value in [ + "windows/amd64", + "linux//v7", + "linux/arm/", + "linux/arm/v7/extra", + ] { + assert!(!super::linux_platform(value), "{value}"); + } + } + #[test] fn oversized_messages_are_rejected_before_sending() { assert!(super::encode(&"x".repeat(super::FRAME_LIMIT)).is_err()); diff --git a/crates/ployz-build/src/tests.rs b/crates/ployz-build/src/tests.rs index 66e8a4f277..5a1d603d6f 100644 --- a/crates/ployz-build/src/tests.rs +++ b/crates/ployz-build/src/tests.rs @@ -189,6 +189,7 @@ exit 0 let targets = [target("api", None), target("web", None)]; let environment = BTreeMap::new(); let request = Request { + image_contexts: &BTreeMap::new(), railpack: &[], compose_file: &directory.join("compose.json"), working_dir: &directory, @@ -284,7 +285,9 @@ fn requested_output_selects_exclusive_bake_behavior() { let planned = plan(&targets).unwrap(); let metadata = Path::new("/private/build-metadata.json"); let build_args = ["MODE=release".to_owned()]; + let image_contexts = BTreeMap::new(); let request = |output| Request { + image_contexts: &image_contexts, railpack: &[], compose_file: Path::new("/private/compose.yaml"), working_dir: Path::new("/private"), diff --git a/crates/ployz-build/src/upload.rs b/crates/ployz-build/src/upload.rs index 94300c2322..694e236acc 100644 --- a/crates/ployz-build/src/upload.rs +++ b/crates/ployz-build/src/upload.rs @@ -255,6 +255,7 @@ impl CompletedUpload { ]); let result = crate::execute_admitted( &Request { + image_contexts: &definition.image_contexts, railpack: &[], compose_file: Path::new("compose.yaml"), working_dir: &upload.root, diff --git a/crates/ployz/src/cli/mod.rs b/crates/ployz/src/cli/mod.rs index c26c510d52..8252f88e45 100644 --- a/crates/ployz/src/cli/mod.rs +++ b/crates/ployz/src/cli/mod.rs @@ -139,17 +139,19 @@ fn project_name(short: Option) -> Arg { value("project-name", short).env(env::COMPOSE_PROJECT_NAME) } +fn build_remote() -> Arg { + Arg::new("remote") + .long("remote") + .num_args(0..=1) + .require_equals(true) + .default_missing_value("") + .value_name("MACHINE") + .conflicts_with("local") +} + fn build() -> Command { base("build", "Build service images") - .arg( - Arg::new("remote") - .long("remote") - .num_args(0..=1) - .require_equals(true) - .default_missing_value("") - .value_name("MACHINE") - .conflicts_with("local"), - ) + .arg(build_remote()) .arg(switch("local", None)) .arg(repeated("build-arg")) .arg(switch("check", None)) @@ -166,6 +168,8 @@ fn build() -> Command { fn deploy() -> Command { base("deploy", "Deploy services from a Compose file") + .arg(build_remote().conflicts_with("no-build")) + .arg(switch("local", None)) .arg(repeated("build-arg")) .arg(switch("build-pull", None)) .arg(many("file", Some('f')).default_value("compose.yaml")) diff --git a/crates/ployz/src/compose/build.rs b/crates/ployz/src/compose/build.rs index 4a3507adcb..2ee89652be 100644 --- a/crates/ployz/src/compose/build.rs +++ b/crates/ployz/src/compose/build.rs @@ -10,6 +10,9 @@ use serde_norway::Value; use super::{BuildSpec, ComposeError, ComposeProject, LoadOptions, build_inputs::BuildInputs}; +#[path = "remote_steps.rs"] +mod remote_steps; + #[derive(Clone, Debug, Default, Eq, PartialEq)] pub struct BuildOptions { pub build_args: Vec, @@ -33,18 +36,16 @@ pub fn plan_build( project: &ComposeProject, options: &BuildOptions, ) -> Result, ComposeError> { - if options.services.is_empty() { - return project - .builds - .keys() - .map(|name| build_service(project, name)) - .collect(); - } + let names = if options.services.is_empty() { + project.builds.keys().cloned().collect::>() + } else { + options.services.clone() + }; let mut selected = Vec::new(); let mut seen = BTreeSet::new(); let mut visiting = BTreeSet::new(); - for name in &options.services { + for name in &names { include_service( project, name, @@ -71,9 +72,24 @@ pub struct CapturedBuild { inputs: BuildInputs, } +/// Where a completed image is available; Machine identity is already resolved. +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] +pub enum BuildLocation { + /// The invoking client’s Docker store contains the completed image. + #[default] + Local, + /// This resolved Machine contains the completed image. + Machine(ployz_core::MachineId), +} + /// A Service whose image this command built, bound to the content produced. #[derive(Clone, Debug, PartialEq)] pub struct BuiltService { + /// Raw Compose Build target name, including build-only dependencies. + /// Build names may contain dots/underscores, unlike Deploy Service names. + pub name: String, + /// Store that holds this completed image. + pub location: BuildLocation, /// Reference the Service requested, used when the image is published. pub image: String, /// Machines this Service is placed on. @@ -173,6 +189,26 @@ pub fn capture_build( } targets.push(ployz_build::Target { name, platform }); retain_service_image_tag(&service.name, image, build)?; + if options.output == Output::Load { + // Bake can import multiple Services under one requested tag in any + // order. Retain every result before a sibling or later client moves it. + let reference = service + .image + .parse::() + .map_err(|error| invalid_build(&error.to_string()))?; + let retained = format!( + "{}/{}:ployz-build-{}", + reference.registry(), + reference.repository(), + uuid::Uuid::new_v4() + ); + let tags = build + .entry(Value::String("tags".into())) + .or_insert_with(|| Value::Sequence(vec![Value::String(service.image.clone())])); + tags.as_sequence_mut() + .ok_or_else(|| invalid_build("invalid build tags"))? + .push(Value::String(retained)); + } if let Some(ssh) = build .get_mut(Value::String("ssh".into())) .and_then(Value::as_sequence_mut) @@ -526,6 +562,23 @@ impl CapturedBuild { cancellation: tokio_util::sync::CancellationToken, progress: impl Fn(ployz_build::Progress), ) -> ployz_build::remote::Outcome { + if self.options.output == Output::Load { + let locations = self + .targets + .iter() + .map(|target| (target.name.clone(), machine_id)) + .collect(); + return match self + .execute_remote_steps(client, &locations, cancellation, progress) + .await + { + Ok(images) => ployz_build::remote::Outcome::Images { + machine_id, + images: images.into_iter().map(|service| service.built).collect(), + }, + Err(outcome) => outcome, + }; + } if !self.railpack.is_empty() { return ployz_build::remote::Outcome::Failed { stage: ployz_build::Stage::Preparation, @@ -534,6 +587,7 @@ impl CapturedBuild { }; } let definition = ployz_build::remote::Definition { + image_contexts: Default::default(), targets: self.targets, output: self.options.output, no_cache: self.options.no_cache, @@ -550,6 +604,25 @@ impl CapturedBuild { .await } + /// Execute remotely and bind each completed output to its captured Service. + /// # Errors + /// Failed, unknown, or image-less outcomes cannot be used by Deploy. + pub async fn execute_remote_images( + self, + client: &crate::connect::Client, + machine_id: ployz_core::MachineId, + cancellation: tokio_util::sync::CancellationToken, + progress: impl Fn(ployz_build::Progress), + ) -> Result, ComposeError> { + let locations = self + .targets + .iter() + .map(|target| (target.name.clone(), machine_id)) + .collect(); + self.execute_on_machines(client, &locations, cancellation, progress) + .await + } + /// Build this capture through the shared runner, without reading the /// original sources again. /// @@ -575,6 +648,7 @@ impl CapturedBuild { .into_owned(), ); let images = ployz_build::execute(&ployz_build::Request { + image_contexts: &BTreeMap::new(), compose_file: Path::new("compose.yaml"), working_dir: self.inputs.root(), environment: &environment, @@ -602,6 +676,8 @@ impl CapturedBuild { .iter() .zip(images) .map(|(service, built)| BuiltService { + name: service.name.clone(), + location: BuildLocation::Local, image: service.image.clone(), machines: service.machines.clone(), built, @@ -823,7 +899,6 @@ fn include_service<'a>( return Ok(()); } visiting.insert(name); - selected.push(name); if let Some(build) = project.builds.get(name) { for dependency in build.additional_services() { include_service(project, dependency, deps, visiting, seen, selected)?; @@ -841,6 +916,7 @@ fn include_service<'a>( )?; } } + selected.push(name); visiting.remove(name); seen.insert(name); Ok(()) diff --git a/crates/ployz/src/compose/build_inputs.rs b/crates/ployz/src/compose/build_inputs.rs index 8ae91c6578..164e05af71 100644 --- a/crates/ployz/src/compose/build_inputs.rs +++ b/crates/ployz/src/compose/build_inputs.rs @@ -124,6 +124,42 @@ impl BuildInputs { }) } + /// Isolate one target's generated recipe while retaining captured bytes. + pub(super) fn for_service(&self, name: &str) -> Result { + let inputs = Self::new()?; + // ponytail: copy the already-filtered capture per remote target. Select + // per-target source trees if large multi-Service captures become costly. + for area in ["source", "private"] { + let source = self.root.join(area); + for entry in entries(&source, &source, None).map_err(input_error)? { + copy( + &entry, + &inputs + .root + .join(area) + .join(entry.file_name().expect("entry")), + &source, + None, + ) + .map_err(input_error)?; + } + } + let mut recipe: serde_norway::Value = serde_norway::from_slice( + &fs::read(self.root.join("compose.yaml")).map_err(input_error)?, + ) + .map_err(|error| ComposeError::Invalid(error.to_string()))?; + recipe + .get_mut("services") + .and_then(serde_norway::Value::as_mapping_mut) + .ok_or_else(|| ComposeError::Invalid("captured Build has no services".into()))? + .retain(|key, _| key.as_str() == Some(name)); + inputs.compose( + &serde_norway::to_string(&recipe) + .map_err(|error| ComposeError::Invalid(error.to_string()))?, + )?; + Ok(inputs) + } + /// Copy a source once, rejecting edits observed during the copy. /// /// # Errors diff --git a/crates/ployz/src/compose/capture.rs b/crates/ployz/src/compose/capture.rs index d473c662b1..7a00fec185 100644 --- a/crates/ployz/src/compose/capture.rs +++ b/crates/ployz/src/compose/capture.rs @@ -30,6 +30,20 @@ pub struct CapturedCompose { } impl CapturedCompose { + /// Use each Service's completed content for Containers and hooks. A tag + /// overwritten by another Service or client cannot substitute its image. + pub fn bind_builds(&mut self, builds: &[super::BuiltService]) { + for service in &mut self.intent.target { + if let Some(build) = builds + .iter() + .find(|build| build.name == service.name.as_str()) + { + service.container.image.clone_from(&build.built.reference); + service.container.pull_policy = ployz_core::PullPolicy::Never; + } + } + } + /// Identity of this capture, unchanged by later edits to the source files. #[must_use] pub fn id(&self) -> &str { diff --git a/crates/ployz/src/compose/mod.rs b/crates/ployz/src/compose/mod.rs index 92c5321b88..833644c3dd 100644 --- a/crates/ployz/src/compose/mod.rs +++ b/crates/ployz/src/compose/mod.rs @@ -9,8 +9,8 @@ mod remote_build; mod secrets; pub use build::{ - BuildOptions, BuildService, BuiltService, CapturedBuild, capture_build, execute_build, - plan_build, + BuildLocation, BuildOptions, BuildService, BuiltService, CapturedBuild, capture_build, + execute_build, plan_build, }; pub use capture::{CapturedCompose, ComposeSource}; pub(crate) use convert::duration_millis; diff --git a/crates/ployz/src/compose/remote_build.rs b/crates/ployz/src/compose/remote_build.rs index e665ac7a19..d91963cc61 100644 --- a/crates/ployz/src/compose/remote_build.rs +++ b/crates/ployz/src/compose/remote_build.rs @@ -22,6 +22,9 @@ pub(super) async fn execute( progress: impl Fn(Progress), ) -> Outcome { let mut evidence = ployz_build::WorkEvidence::new(&definition.targets); + if cancellation.is_cancelled() { + return failed(Stage::Admission, "Build cancelled before submission").with_work(evidence); + } if let Err(error) = remote::validate_capture(inputs.root(), &definition) { return failed(Stage::Preparation, error.to_string()).with_work(evidence); } @@ -174,7 +177,7 @@ fn validate_outcome( .and_then(|reference| reference.digest().map(str::to_owned)) .is_none() || image.tags.is_empty() - || !linux_platform(&image.platform) + || !remote::linux_platform(&image.platform) }) { return unknown( Stage::Output, @@ -204,19 +207,6 @@ fn validate_outcome( outcome } -fn linux_platform(platform: &str) -> bool { - let valid = |part: &str| { - !part.is_empty() - && part - .bytes() - .all(|byte| byte.is_ascii_alphanumeric() || b"._-".contains(&byte)) - }; - let mut parts = platform.split('/'); - parts.next() == Some("linux") - && parts.next().is_some_and(valid) - && parts.next().is_none_or(valid) - && parts.next().is_none() -} fn failed(stage: Stage, message: impl Into) -> Outcome { Outcome::Failed { work: Default::default(), diff --git a/crates/ployz/src/compose/remote_steps.rs b/crates/ployz/src/compose/remote_steps.rs new file mode 100644 index 0000000000..8c13d100a4 --- /dev/null +++ b/crates/ployz/src/compose/remote_steps.rs @@ -0,0 +1,180 @@ +//! Execute a captured remote plan in dependency order. A consumer gets the +//! completed dependency's immutable image served by its actual Build Machine. + +use super::*; +use ployz_build::{ + Progress, Stage, TargetEvidence, WorkEvidence, + remote::{Definition, Outcome}, +}; +use ployz_core::MachineId; +use tokio_util::sync::CancellationToken; + +impl CapturedBuild { + /// Build on already-resolved Machines, independently of application placement. + /// Every target must have a location before any source is submitted. + /// Keys are raw Compose Build names, not DNS-label Deploy Service names. + /// # Errors + /// Reports failed/unknown work and leaves dependent/remaining targets unattempted. + pub async fn execute_on_machines( + self, + client: &crate::connect::Client, + locations: &BTreeMap, + cancellation: CancellationToken, + progress: impl Fn(Progress), + ) -> Result, ComposeError> { + self.execute_remote_steps(client, locations, cancellation, progress) + .await + .map_err(remote_error) + } + + pub(super) async fn execute_remote_steps( + self, + client: &crate::connect::Client, + locations: &BTreeMap, + cancellation: CancellationToken, + progress: impl Fn(Progress), + ) -> Result, Outcome> { + let mut work = WorkEvidence::new(&self.targets); + let failed = |stage, message| Outcome::Failed { + stage, + message, + work: Default::default(), + }; + if self.options.output != Output::Load { + return Err(failed( + Stage::Preparation, + "per-Machine execution requires loaded image results".into(), + ) + .with_work(work)); + } + if !self.railpack.is_empty() { + return Err(failed(Stage::Preparation, "remote builds currently support Dockerfile recipes only; Railpack requires a local build".into()).with_work(work)); + } + if self + .targets + .iter() + .any(|target| !locations.contains_key(&target.name)) + { + return Err(failed( + Stage::Admission, + "every Build must have a selected Machine before execution".into(), + ) + .with_work(work)); + } + let mut completed: Vec = Vec::new(); + for (service, target) in self.plan.iter().zip(&self.targets) { + if cancellation.is_cancelled() { + return Err(failed( + Stage::Admission, + "Build cancelled before the next target was submitted".into(), + ) + .with_work(work)); + } + let machine_id = *locations.get(&target.name).expect("all locations checked"); + let build = super::BuildSpec { + raw: service.build.clone(), + }; + let mut contexts = BTreeMap::new(); + for dependency in build.additional_services() { + let Some(image) = completed.iter().find(|image| image.name == dependency) else { + return Err(failed( + Stage::Preparation, + format!("Service context {dependency} has no completed Build output"), + ) + .with_work(work)); + }; + let BuildLocation::Machine(source) = image.location else { + unreachable!("remote steps return Machine images") + }; + let serving = + crate::image::serve_build_image(&mut client.clone(), &image.built, source) + .await; + let source = serving.map_err(|error| { + failed( + Stage::Preparation, + format!("Service context {dependency}: {error}"), + ) + .with_work(work.clone()) + })?; + contexts.insert( + dependency.to_owned(), + ployz_build::ImageContext { + reference: image.built.reference.clone(), + platform: image.built.platform.clone(), + source, + }, + ); + } + let inputs = self.inputs.for_service(&target.name).map_err(|error| { + failed(Stage::Preparation, error.to_string()).with_work(work.clone()) + })?; + let outcome = super::super::remote_build::execute( + inputs, + Definition { + targets: vec![target.clone()], + image_contexts: contexts, + output: Output::Load, + no_cache: self.options.no_cache, + pull: self.options.pull, + }, + client, + machine_id, + cancellation.clone(), + &progress, + ) + .await; + match outcome { + Outcome::Images { images, .. } => { + let image = images + .into_iter() + .next() + .expect("remote adapter validated the count"); + work.0 + .insert(target.name.clone(), TargetEvidence::Image(image.clone())); + completed.push(BuiltService { + name: service.name.clone(), + image: service.image.clone(), + machines: service.machines.clone(), + location: BuildLocation::Machine(machine_id), + built: image, + }); + } + outcome @ (Outcome::Failed { .. } | Outcome::Unknown { .. }) => { + if let Outcome::Failed { work: observed, .. } + | Outcome::Unknown { work: observed, .. } = &outcome + && let Some(evidence) = observed.0.get(&target.name) + { + work.0.insert(target.name.clone(), evidence.clone()); + } + return Err(outcome.with_work(work)); + } + Outcome::Validated { .. } | Outcome::Published { .. } => { + unreachable!("adapter validated output disposition") + } + } + } + Ok(completed) + } +} + +pub(super) fn remote_error(outcome: Outcome) -> ComposeError { + match outcome { + Outcome::Failed { + stage, + message, + work, + } => invalid_build(&format!( + "Build failed during {stage:?}: {message}; target evidence: {work:?}" + )), + Outcome::Unknown { + stage, + message, + work, + } => invalid_build(&format!( + "Build outcome unknown during {stage:?}: {message}; target evidence: {work:?}" + )), + Outcome::Images { .. } | Outcome::Validated { .. } | Outcome::Published { .. } => { + invalid_build("Build produced no image available for Direct Image Transfer") + } + } +} diff --git a/crates/ployz/src/deploy/apply.rs b/crates/ployz/src/deploy/apply.rs index 8ddce24740..0d391d80a2 100644 --- a/crates/ployz/src/deploy/apply.rs +++ b/crates/ployz/src/deploy/apply.rs @@ -158,7 +158,8 @@ pub(crate) async fn deploy_project( gate: ConfirmGate<'_>, ) -> Result<(), Failure> { let machines = client.machines().await?; - let outcome = push_project_images(client, builds, &machines).await?; + let preview = plan_project(client, candidate, machines.clone()).await?; + let outcome = push_project_images(client, builds, &machines, &preview).await?; print_pushed_images(&outcome); if !outcome.failures.is_empty() { return Err(Failure::usage(format!( @@ -166,7 +167,6 @@ pub(crate) async fn deploy_project( outcome.failures.join("; ") ))); } - let preview = plan_project(client, candidate, machines).await?; println!("Captured candidate {}", candidate.id()); print_warnings(&preview); confirm_and_execute(client, &preview, gate).await diff --git a/crates/ployz/src/deploy/pipeline.rs b/crates/ployz/src/deploy/pipeline.rs index 2ad3613c5b..77035d1176 100644 --- a/crates/ployz/src/deploy/pipeline.rs +++ b/crates/ployz/src/deploy/pipeline.rs @@ -281,11 +281,34 @@ pub(super) async fn push_project_images( client: &mut Client, builds: &[BuiltService], machines: &[MachineObservation], + preview: &DeployPlan, ) -> Result { let mut pushed = Vec::new(); let mut failures = Vec::new(); - for service in builds { - match push_image(client, service, machines).await { + // Check every actual destination before any image or application changes. + let deliveries = builds.iter().map(|service| { + let targets = preview.operations.iter() + .filter(|row| row.operation.spec().is_some_and(|spec| spec.name.as_str() == service.name)) + .map(|row| row.machine_id).collect::>(); + for target in &targets { + let architecture = machines.iter().find(|machine| machine.machine.id == *target) + .map(|machine| machine.machine.runtime.architecture.as_str()); + let compatible = match architecture { + Some("x86_64" | "amd64") => service.built.platform == "linux/amd64", + Some("aarch64" | "arm64") => matches!(service.built.platform.as_str(), "linux/arm64" | "linux/arm64/v8"), + _ => false, + }; + if !compatible { + return Err(Failure::usage(format!("Build for Service {} contains {}; destination Machine {target} reports architecture {}. No Service, hook, or volume change was attempted.", service.name, service.built.platform, architecture.unwrap_or("unknown")))); + } + } + Ok((service, targets.into_iter().map(|target| target.to_string()).collect::>())) + }).collect::, Failure>>()?; + for (service, targets) in deliveries { + if targets.is_empty() { + continue; + } + match push_image(client, service, machines, &targets).await { Ok((images, service_failures)) => { pushed.extend(images); failures.extend(service_failures); @@ -528,16 +551,35 @@ async fn push_image( client: &mut Client, service: &BuiltService, machines: &[MachineObservation], + targets: &[String], ) -> Result<(Vec, Vec), PushError> { - let targets = service - .machines - .iter() - .map(ToString::to_string) - .collect::>(); - // Deliver the content this command built, not whatever the tag now holds. - let result = - crate::image::push_using_machines(client, service.content(), None, &targets, machines) - .await?; + let result = match service.location { + crate::compose::BuildLocation::Local => { + // Retain each digest even when multiple Services requested one tag. + let tag = service + .built + .reference + .replace("@sha256:", ":ployz-sha256-"); + crate::image::push_using_machines( + client, + crate::image::ImageContent::built(&tag, &service.built.reference), + Some(&service.built.platform), + targets, + machines, + ) + .await? + } + crate::compose::BuildLocation::Machine(source) => { + crate::image::push_from_machine_using_machines( + client, + &service.built, + source, + targets, + machines, + ) + .await? + } + }; let pushed = result .successes .iter() diff --git a/crates/ployz/src/handlers/build.rs b/crates/ployz/src/handlers/build.rs index 5003347bcc..30e29a8b3a 100644 --- a/crates/ployz/src/handlers/build.rs +++ b/crates/ployz/src/handlers/build.rs @@ -58,34 +58,10 @@ pub(super) fn run(matches: &ArgMatches) -> Result<(), Error> { return runtime()?.block_on(async { let cancellation = super::cancellation_on_ctrl_c(); let mut client = connect_client(matches, context.as_deref()).await?; - let machine = client.build_machine(&target).await?; - println!("Selected Build Machine {} ({})", machine.name, machine.id); - let contract = client - .invoke::( - ployz_core::DescribeContractRequest {}, - &ployz_core::MachineTarget::from(&machine.id), - Some(std::time::Duration::from_secs(5)), - ) - .await?; - if contract.machine_id != machine.id || !contract.supports(ployz_core::BUILD_CAPABILITY) - { - return Err(Error::usage(format!( - "Machine {} does not support remote Builds", - machine.id - ))); - } + let machine = select_build_machine(&mut client, &target).await?; let captured = capture_build(&plan, &options, &mut project)?; let outcome = captured - .execute_remote(&client, machine.id, cancellation.clone(), |event| { - use std::io::Write as _; - match event { - ployz_build::Progress::Stage(stage) => eprintln!("Build: {stage:?}"), - ployz_build::Progress::Target { .. } => {} - ployz_build::Progress::Output(bytes) => { - let _ = std::io::stderr().write_all(&bytes); - } - } - }) + .execute_remote(&client, machine.id, cancellation.clone(), progress) .await; cancellation.cancel(); report_remote(outcome) @@ -141,6 +117,39 @@ pub(super) fn run(matches: &ArgMatches) -> Result<(), Error> { } } +pub(super) async fn select_build_machine( + client: &mut crate::connect::Client, + target: &ployz_core::MachineTarget, +) -> Result { + let machine = client.build_machine(target).await?; + println!("Selected Build Machine {} ({})", machine.name, machine.id); + let contract = client + .invoke::( + ployz_core::DescribeContractRequest {}, + &ployz_core::MachineTarget::from(&machine.id), + Some(std::time::Duration::from_secs(5)), + ) + .await?; + if contract.machine_id != machine.id || !contract.supports(ployz_core::BUILD_CAPABILITY) { + return Err(Error::usage(format!( + "Machine {} does not support remote Builds", + machine.id + ))); + } + Ok(machine) +} + +pub(super) fn progress(event: ployz_build::Progress) { + use std::io::Write as _; + match event { + ployz_build::Progress::Stage(stage) => eprintln!("Build: {stage:?}"), + ployz_build::Progress::Target { .. } => {} + ployz_build::Progress::Output(bytes) => { + let _ = std::io::stderr().write_all(&bytes); + } + } +} + /// Validation supersedes publication: a checked recipe produces no image. fn requested_output(leaf: &ArgMatches) -> Output { if leaf.get_flag("check") { diff --git a/crates/ployz/src/handlers/deploy.rs b/crates/ployz/src/handlers/deploy.rs index 2e3f3644da..1944bad57c 100644 --- a/crates/ployz/src/handlers/deploy.rs +++ b/crates/ployz/src/handlers/deploy.rs @@ -5,7 +5,7 @@ use ployz_core::{ComposePruneRefusal, ServiceSelector}; use crate::{ compose::{ - BuildOptions, BuiltService, CapturedCompose, ComposeError, ComposeProject, LoadOptions, + BuildOptions, CapturedBuild, CapturedCompose, ComposeError, ComposeProject, LoadOptions, capture_build, compose_identity, has_explicit_nondefault_compose_file, load_project, plan_build, }, @@ -42,6 +42,13 @@ pub(super) fn run(root: &ArgMatches) -> Result<(), Error> { pub(super) fn deploy(root: &ArgMatches) -> Result<(), Error> { let matches = leaf_matches(root); + let remote = matches.get_one::("remote"); + if remote.is_some_and(String::is_empty) { + return Err(Error::usage( + "select a Build Machine with --remote=; automatic selection is not available", + )); + } + let remote = remote.map(ployz_core::MachineTarget::parse).transpose()?; let load = deploy_load(matches); let resolved = resolve_from_compose_load(matches, &load)?; let project = load_project(&load)?; @@ -56,9 +63,40 @@ pub(super) fn deploy(root: &ArgMatches) -> Result<(), Error> { let skip_health_monitor = matches.get_flag("skip-health"); let mut options = plan_options(force_recreate, skip_health_monitor); options.selected = selected_attempts(&project, &string_values(matches, "service"))?; - let (candidate, builds) = prepare_deploy(matches, &load, project, &resolved, options)?; + let (mut candidate, captured_build) = + prepare_deploy(matches, &load, project, &resolved, options)?; runtime()?.block_on(async { let mut client = connect_client(root, context.as_deref()).await?; + // Every required Build finishes before preparation or application changes. + let builds = match captured_build { + Some(build) => match remote { + Some(target) => { + let machine = super::build::select_build_machine(&mut client, &target).await?; + let cancellation = super::cancellation_on_ctrl_c(); + let result = build + .execute_remote_images( + &client, + machine.id, + cancellation.clone(), + super::build::progress, + ) + .await; + let cancelled = cancellation.is_cancelled(); + cancellation.cancel(); + if cancelled { + return Err(Error::usage( + "Build cancelled. No Service, hook, or volume change was attempted.", + )); + } + result.map_err(crate::deploy::DeployError::from)? + } + None => build + .execute(load.docker.as_deref()) + .map_err(crate::deploy::DeployError::from)?, + }, + None => Vec::new(), + }; + candidate.bind_builds(&builds); deploy_project( &mut client, &candidate, @@ -257,7 +295,7 @@ fn prepare_deploy( mut project: ComposeProject, resolved: &ResolvedProject, options: ployz_core::PlanOptions, -) -> Result<(CapturedCompose, Vec), Error> { +) -> Result<(CapturedCompose, Option), Error> { let selected = string_values(matches, "service"); for warning in &project.warnings { eprintln!("WARNING: {warning}"); @@ -271,13 +309,13 @@ fn prepare_deploy( build_args: string_values(matches, "build-arg"), deps: true, no_cache: matches.get_flag("no-cache"), - // A Deploy needs the image on this host, so it always loads it. + // A Deploy needs the image in its execution host’s image store. output: ployz_build::Output::Load, pull: matches.get_flag("build-pull"), services: build_names, }; let builds = plan_build(&project, &build_options)?; - let captured_build = if matches.get_flag("no-build") { + let captured_build = if matches.get_flag("no-build") || builds.is_empty() { None } else { Some(capture_build(&builds, &build_options, &mut project)?) @@ -291,14 +329,7 @@ fn prepare_deploy( hints.compose_refusal, load.files.clone(), ); - // Every required Build finishes before any application change begins. - let built = match captured_build { - Some(build) => build - .execute(load.docker.as_deref()) - .map_err(crate::deploy::DeployError::from)?, - None => Vec::new(), - }; - Ok((candidate, built)) + Ok((candidate, captured_build)) } fn selected_attempts( @@ -358,3 +389,7 @@ use input::{parse_u32, run_spec}; #[cfg(test)] #[path = "deploy_tests.rs"] mod tests; + +#[cfg(test)] +#[path = "deploy_remote_tests.rs"] +mod remote_tests; diff --git a/crates/ployz/src/handlers/deploy_remote_tests.rs b/crates/ployz/src/handlers/deploy_remote_tests.rs new file mode 100644 index 0000000000..bf1ad46680 --- /dev/null +++ b/crates/ployz/src/handlers/deploy_remote_tests.rs @@ -0,0 +1,354 @@ +//! Rung 2: the product handler against owned Machine transport, with sentinels +//! for every application mutation. Docker execution is proved at rung 4. +use ployz_build::{Stage, TargetEvidence, WorkEvidence, remote::Outcome}; +use ployz_core::{MachineImages, MembershipObservation, RpcError, RpcErrorCode}; +use std::{ + fs, + path::{Path, PathBuf}, + sync::{Arc, atomic::Ordering}, +}; + +#[path = "../../tests/deploy_client/support.rs"] +#[allow(dead_code)] +mod support; +use support::*; + +fn fixture() -> (PathBuf, DeployService, Arc) { + let root = std::env::temp_dir().join(format!("ployz-deploy-803-{}", uuid::Uuid::new_v4())); + fs::create_dir_all(&root).unwrap(); + fs::write(root.join("Dockerfile"), "FROM scratch\n").unwrap(); + let mut source = machine('a', "builder"); + let mut destination = machine('b', "application"); + source.machine.runtime.architecture = "x86_64".into(); + destination.machine.runtime.architecture = "x86_64".into(); + fs::write(root.join("compose.yaml"), format!( + "name: example\nservices:\n one:\n image: registry.invalid/shared:latest\n build: .\n x-machines: [{}]\n x-pre_deploy: {{command: ['true']}}\n volumes: [data:/data]\n two:\n image: registry.invalid/shared:latest\n build: .\n x-machines: [{}]\nvolumes:\n data: {{}}\n", + destination.machine.id, destination.machine.id, + )).unwrap(); + let builds = Arc::new(BuildFixture::default()); + let mut service = DeployService::new(source.clone()) + .with_machines(vec![source, destination]) + .with_exec_exit(0); + service.builds = Some(builds.clone()); + (root, service, builds) +} + +async fn deploy(root: &Path, service: DeployService) -> Result<(), crate::failure::Failure> { + let (address, server) = listening(service).await; + let config = root.join("config.yaml"); + let file = root.join("compose.yaml"); + let matches = crate::cli::command() + .try_get_matches_from([ + "ployz", + "--connect", + &format!("tcp://{address}"), + "--ployz-config", + config.to_str().unwrap(), + "deploy", + "--remote=builder", + "--yes", + "--skip-health", + "--file", + file.to_str().unwrap(), + ]) + .unwrap(); + let result = tokio::task::spawn_blocking(move || super::deploy(&matches)) + .await + .unwrap(); + server.abort(); + result +} + +#[tokio::test] +async fn remote_deploy_uses_each_services_completed_content_without_local_inspection() { + let (root, service, builds) = fixture(); + let created = service.created_specs(); + let yaml = fs::read_to_string(root.join("compose.yaml")).unwrap(); + fs::write( + root.join("compose.yaml"), + yaml.replace(" x-pre_deploy: {command: ['true']}\n", ""), + ) + .unwrap(); + deploy(&root, service).await.unwrap(); + let created = created.lock().unwrap(); + assert!(!created.is_empty()); + for spec in created.iter() { + let digest = if spec.name.as_str() == "one" { + "1" + } else { + "2" + } + .repeat(64); + assert_eq!( + spec.container.image, + format!("registry.invalid/shared@sha256:{digest}") + ); + assert_eq!(spec.container.pull_policy, ployz_core::PullPolicy::Never); + } + let pulls = builds.pulls.lock().unwrap(); + assert_eq!(pulls.len(), 2); + assert!(pulls.iter().all( + |(target, pull)| *target == machine('b', "application").machine.id + && pull.image.contains("@sha256:") + && pull.source.management_address + == machine('a', "builder").machine.management_address() + )); + assert!( + builds + .opened + .lock() + .unwrap() + .iter() + .all(|id| *id == machine('a', "builder").machine.id) + ); + fs::remove_dir_all(root).unwrap(); +} + +#[tokio::test] +async fn failed_unknown_and_cancelled_builds_leave_hooks_containers_and_volumes_unattempted() { + for failure in ["failed", "cancelled", "unknown"] { + let (root, service, builds) = fixture(); + let mutations = service.mutating_rpcs(); + let work = WorkEvidence(std::collections::BTreeMap::from([ + ("one".into(), TargetEvidence::Unknown), + ("two".into(), TargetEvidence::Unattempted), + ])); + *builds.terminal.lock().unwrap() = Some(if failure == "unknown" { + Outcome::Unknown { + stage: Stage::Building, + message: "termination unconfirmed".into(), + work, + } + } else { + Outcome::Failed { + stage: Stage::Building, + message: format!("Build {failure}"), + work, + } + }); + let error = deploy(&root, service).await.unwrap_err().to_string(); + assert!( + error.contains("No Service, hook, or volume change was attempted"), + "{error}" + ); + assert!(error.contains("Unattempted"), "{error}"); + assert_eq!(mutations.load(Ordering::SeqCst), 0); + assert!(builds.pulls.lock().unwrap().is_empty()); + assert!(builds.opened.lock().unwrap().is_empty()); + fs::remove_dir_all(root).unwrap(); + } +} + +#[tokio::test] +async fn incompatible_application_platform_refuses_before_transfer_or_mutations() { + let (root, service, builds) = fixture(); + let mutations = service.mutating_rpcs(); + let mut destination = machine('b', "application"); + destination.machine.runtime.architecture = "aarch64".into(); + let service = service.with_machines(vec![machine('a', "builder"), destination]); + let error = deploy(&root, service).await.unwrap_err().to_string(); + assert!( + error.contains("linux/amd64") && error.contains("aarch64"), + "{error}" + ); + assert_eq!(mutations.load(Ordering::SeqCst), 0); + assert!(builds.pulls.lock().unwrap().is_empty()); + fs::remove_dir_all(root).unwrap(); +} + +#[tokio::test] +async fn remote_transfer_keeps_exact_source_successes_failures_and_omissions() { + let (root, service, builds) = fixture(); + let source = machine('a', "builder"); + let failed = machine('b', "failed"); + let success = machine('c', "success"); + let missing = ployz_core::MachineObservation::new( + machine('d', "missing").machine, + MembershipObservation::Down, + ); + let image = ployz_build::BuiltImage { + reference: format!("registry.invalid/shared@sha256:{}", "1".repeat(64)), + tags: vec!["registry.invalid/shared:latest".into()], + platform: "linux/amd64".into(), + }; + builds.stores.lock().unwrap().insert( + source.machine.id, + MachineImages { + containerd_store: true, + images: vec![ployz_core::ImageSummary { + id: format!("sha256:{}", "1".repeat(64)), + // The requested tag now points elsewhere; only content proves identity. + repo_tags: vec!["registry.invalid/shared:retained".into()], + created: 0, + size: 1, + containers: 0, + platforms: vec!["linux/amd64".into()], + }], + }, + ); + builds.pull_failures.lock().unwrap().insert( + failed.machine.id, + RpcError { + code: RpcErrorCode::Unsupported, + message: "containerd image store unavailable".into(), + details: serde_json::Value::Null, + }, + ); + let (mut client, server) = connected(service.with_machines(vec![ + source.clone(), + failed.clone(), + success.clone(), + missing.clone(), + ])) + .await; + let result = crate::image::push_from_machine(&mut client, &image, source.machine.id, &[]) + .await + .unwrap(); + assert_eq!( + result + .successes + .iter() + .map(|success| success.machine_id) + .collect::>(), + [source.machine.id, success.machine.id] + ); + assert_eq!(result.failures.len(), 1); + assert_eq!( + result.failures.first().unwrap().machine_id, + failed.machine.id + ); + assert!( + result + .failures + .first() + .unwrap() + .error + .to_string() + .contains("containerd image store") + ); + assert_eq!(result.omissions, [missing.machine.id]); + assert!( + builds + .pulls + .lock() + .unwrap() + .iter() + .all(|(_, pull)| pull.image == image.reference + && pull.source.management_address == source.machine.management_address()) + ); + builds + .stores + .lock() + .unwrap() + .get_mut(&source.machine.id) + .unwrap() + .containerd_store = false; + assert!(matches!( + crate::image::push_from_machine( + &mut client, + &image, + source.machine.id, + &[success.machine.id.to_string()] + ) + .await, + Err(crate::image::PushError::UnsupportedImageStore) + )); + server.abort(); + fs::remove_dir_all(root).unwrap(); +} + +#[tokio::test] +async fn service_context_uses_the_completed_dependency_on_a_different_build_machine() { + let (root, service, builds) = fixture(); + let source = machine('a', "builder").machine.id; + let consumer = machine('b', "application").machine.id; + let mut project = crate::compose::parse_normalized( + "services: {base: {image: 'registry.invalid/shared:latest', build: .}, app: {image: 'registry.invalid/shared:latest', build: {context: ., additional_contexts: {base: 'service:base'}}}}", + &root, + ).unwrap(); + let options = crate::compose::BuildOptions { + services: vec!["app".into()], + ..Default::default() + }; + let plan = crate::compose::plan_build(&project, &options).unwrap(); + let captured = crate::compose::capture_build(&plan, &options, &mut project).unwrap(); + // Later edits must not change either Service's admitted recipe or source. + fs::write(root.join("Dockerfile"), "invalid later edit\n").unwrap(); + let (client, server) = connected(service).await; + let images = captured + .execute_on_machines( + &client, + &std::collections::BTreeMap::from([("base".into(), source), ("app".into(), consumer)]), + tokio_util::sync::CancellationToken::new(), + |_| {}, + ) + .await + .unwrap(); + assert_eq!( + images + .iter() + .map(|image| image.name.as_str()) + .collect::>(), + ["base", "app"] + ); + assert_eq!( + images.first().unwrap().location, + crate::compose::BuildLocation::Machine(source) + ); + assert_eq!( + images.get(1).unwrap().location, + crate::compose::BuildLocation::Machine(consumer) + ); + let definitions = builds.definitions.lock().unwrap(); + assert_eq!(definitions.len(), 2); + let dependency = definitions + .get(1) + .unwrap() + .image_contexts + .get("base") + .unwrap(); + assert_eq!( + dependency.reference, + format!("registry.invalid/shared@sha256:{}", "1".repeat(64)) + ); + assert_eq!( + dependency.source.management_address, + machine('a', "builder").machine.management_address() + ); + assert_eq!(builds.opened.lock().unwrap().as_slice(), [source]); + assert!( + builds.pulls.lock().unwrap().is_empty(), + "BuildKit consumes the source directly" + ); + server.abort(); + fs::remove_dir_all(root).unwrap(); +} + +#[tokio::test] +async fn incomplete_build_locations_refuse_before_source_submission() { + let (root, service, builds) = fixture(); + let mut project = + crate::compose::parse_normalized("services: {api: {build: .}}", &root).unwrap(); + let options = crate::compose::BuildOptions::default(); + let plan = crate::compose::plan_build(&project, &options).unwrap(); + let captured = crate::compose::capture_build(&plan, &options, &mut project).unwrap(); + let (client, server) = connected(service).await; + let error = captured + .execute_on_machines( + &client, + &Default::default(), + tokio_util::sync::CancellationToken::new(), + |_| {}, + ) + .await + .unwrap_err() + .to_string(); + assert!( + error.contains("every Build must have a selected Machine"), + "{error}" + ); + assert!(error.contains("Unattempted"), "{error}"); + assert!(builds.definitions.lock().unwrap().is_empty()); + server.abort(); + fs::remove_dir_all(root).unwrap(); +} diff --git a/crates/ployz/src/handlers/deploy_tests.rs b/crates/ployz/src/handlers/deploy_tests.rs index fb37d759e8..56ba2dce79 100644 --- a/crates/ployz/src/handlers/deploy_tests.rs +++ b/crates/ployz/src/handlers/deploy_tests.rs @@ -343,7 +343,7 @@ fn deploy_preparation_captures_resolved_input_and_selection_before_planning() { }; std::fs::write(&source, yaml.replace("captured", "later-edit")).unwrap(); let (candidate, builds) = prepare_deploy(matches, &load, project, &resolved, options).unwrap(); - assert!(builds.is_empty()); + assert!(builds.is_none()); assert_eq!( std::fs::read_to_string(directory.join("calls")).unwrap(), "x" @@ -409,16 +409,18 @@ fn a_failed_build_leaves_the_deployment_unattempted() { source: ProjectNameSource::CommandLine, }; - let error = match prepare_deploy( + let (_, build) = prepare_deploy( leaf_matches(&root), &load, project, &resolved, ployz_core::PlanOptions::default(), - ) { - Ok(_) => panic!("a failed build was admitted for deployment"), - Err(error) => error.to_string(), - }; + ) + .unwrap(); + let error = crate::deploy::DeployError::from( + build.unwrap().execute(load.docker.as_deref()).unwrap_err(), + ) + .to_string(); assert!( error.contains("No Service, hook, or volume change was attempted"), "{error}" diff --git a/crates/ployz/src/image.rs b/crates/ployz/src/image.rs index c234f9ff4d..43e433100f 100644 --- a/crates/ployz/src/image.rs +++ b/crates/ployz/src/image.rs @@ -22,7 +22,10 @@ use crate::{ use self::proxy::{ImageProxy, ProxyMode, detect_mode}; +mod built; mod proxy; +pub use built::push_from_machine; +pub(crate) use built::{push_from_machine_using_machines, serve_build_image}; #[must_use] pub fn with_default_tag(image: &str) -> String { @@ -39,6 +42,14 @@ pub fn with_default_tag(image: &str) -> String { #[derive(Debug, Error)] pub enum PushError { + #[error( + "Build image {image} on Machine {machine_id} is not available with platform {platform}" + )] + BuildImageUnavailable { + image: String, + machine_id: MachineId, + platform: String, + }, #[error("invalid image reference '{reference}': {message}")] InvalidReference { reference: String, message: String }, #[error("direct image push requires a tagged local reference")] diff --git a/crates/ployz/src/image/built.rs b/crates/ployz/src/image/built.rs new file mode 100644 index 0000000000..2b3ae09b7e --- /dev/null +++ b/crates/ployz/src/image/built.rs @@ -0,0 +1,123 @@ +//! Deliver a completed Build from the Machine that holds its exact content. + +use ployz_build::BuiltImage; +use ployz_core::{ListImagesRequest, MachineObservation}; + +use super::*; + +/// Transfer the completed image directly from its Build Machine. The client +/// neither inspects Docker nor carries image bytes or selects another source. +/// # Errors +/// Reports selection, source capability, and source image-store failures. +/// Destination failures and unattempted destinations remain in the partial result. +pub async fn push_from_machine( + client: &mut Client, + image: &BuiltImage, + source: MachineId, + selectors: &[String], +) -> Result, PushError> { + let machines = client.machines().await?; + push_from_machine_using_machines(client, image, source, selectors, &machines).await +} + +pub(crate) async fn push_from_machine_using_machines( + client: &mut Client, + image: &BuiltImage, + source: MachineId, + selectors: &[String], + machines: &[MachineObservation], +) -> Result, PushError> { + let selection = list_selection(machines, selectors)?; + let mut result = PartialResult { + successes: Vec::new(), + failures: Vec::new(), + omissions: selection.omissions, + }; + if selection.targets.is_empty() { + return Ok(result); + } + let mut cancellation = Cancellation::new(); + let source = serve_build_image(client, image, source).await?; + let mut remaining = selection.targets.into_iter(); + while let Some(machine) = remaining.next() { + match pull_on_machine( + client, + &image.reference, + &machine, + source, + &mut cancellation, + ) + .await + { + Ok(()) => result.successes.push(MachineSuccess { + machine_id: machine.id, + value: (), + }), + Err(error) => { + let cancelled = error.is_cancellation(); + result.failures.push(MachineFailure { + machine_id: machine.id, + error, + }); + if cancelled { + result.omissions.extend(remaining.map(|machine| machine.id)); + break; + } + } + } + } + Ok(result) +} + +/// Open the actual complete Build source for either delivery or a later Build. +pub(crate) async fn serve_build_image( + client: &mut Client, + image: &BuiltImage, + source: MachineId, +) -> Result { + let mut cancellation = Cancellation::new(); + let reference = + image + .reference + .parse::() + .map_err(|error| PushError::InvalidReference { + reference: image.reference.clone(), + message: error.to_string(), + })?; + let digest = reference + .digest() + .ok_or_else(|| PushError::InvalidReference { + reference: image.reference.clone(), + message: "a completed Build must have an immutable digest".into(), + })?; + let target = MachineTarget::from(&source); + // Docker's reference filter does not match repository@digest. Compare the + // manifest identity and available platforms in the source's actual store. + let stored = cancellation + .race(client.call::(ListImagesRequest { reference: None }, Some(&target))) + .await??; + if !stored.containerd_store { + return Err(PushError::UnsupportedImageStore); + } + if !stored.images.iter().any(|stored| { + stored.id == digest + && stored.platforms.iter().any(|platform| { + platform == &image.platform + || matches!( + (platform.as_str(), image.platform.as_str()), + ("linux/arm64", "linux/arm64/v8") | ("linux/arm64/v8", "linux/arm64") + ) + }) + }) { + return Err(PushError::BuildImageUnavailable { + image: image.reference.clone(), + machine_id: source, + platform: image.platform.clone(), + }); + } + let opened = cancellation + .race(client.call::(EnsureImageIngestRequest {}, Some(&target))) + .await? + .map_err(|error| ingest_error(rpc_error(error)))?; + Ok(opened.destination) +} diff --git a/crates/ployz/src/lib.rs b/crates/ployz/src/lib.rs index 38a9166450..da961fbba4 100644 --- a/crates/ployz/src/lib.rs +++ b/crates/ployz/src/lib.rs @@ -20,3 +20,6 @@ pub mod sdk; pub mod service; mod setup_retry; pub mod volume; +// Shared Machine transport fixtures also exercise the in-process CLI handlers. +#[cfg(test)] +extern crate self as ployz; diff --git a/crates/ployz/tests/build.rs b/crates/ployz/tests/build.rs index ce6e6c94a5..7f940fe07f 100644 --- a/crates/ployz/tests/build.rs +++ b/crates/ployz/tests/build.rs @@ -53,11 +53,11 @@ services: .iter() .map(|service| service.name.as_str()) .collect::>(), - ["api", "base"] + ["base", "api"] ); - assert_eq!(direct.first().unwrap().image, "example.test/api:version2"); + assert_eq!(direct.get(1).unwrap().image, "example.test/api:version2"); assert_eq!( - direct.get(1).unwrap().image, + direct.first().unwrap().image, project.services.get("base").unwrap().container.image ); @@ -75,7 +75,7 @@ services: .iter() .map(|service| service.name.as_str()) .collect::>(), - ["api", "base", "database", "frontend"] + ["base", "database", "api", "frontend"] ); assert_eq!( with_deps.get(3).unwrap().image, @@ -119,7 +119,7 @@ services: .iter() .map(|service| service.name.as_str()) .collect::>(), - ["api", "base"] + ["base", "api"] ); let cycle = parse_normalized( @@ -139,6 +139,7 @@ services: .to_string() .contains("build dependency cycle") ); + assert!(plan_build(&cycle, &BuildOptions::default()).is_err()); } #[test] @@ -695,3 +696,45 @@ mod capture; #[path = "build/railpack.rs"] mod railpack; + +#[test] +fn deploy_binds_each_service_to_its_build_when_requested_tags_are_shared() { + use ployz::compose::BuildLocation; + let project = parse_normalized( + "services: {one: {image: 'example.test/shared:latest', build: .}, two: {image: 'example.test/shared:latest', build: .}}", + ".", + ).unwrap(); + let mut candidate = project.capture( + ployz_core::ProjectName::parse("app").unwrap(), + Default::default(), + vec![], + None, + vec![], + ); + let builds = + [("one", FIRST_CONTENT), ("two", SECOND_CONTENT)].map(|(name, digest)| BuiltService { + name: name.into(), + image: "example.test/shared:latest".into(), + machines: vec![], + location: BuildLocation::Machine(ployz_core::MachineId::parse("a".repeat(32)).unwrap()), + built: ployz_build::BuiltImage { + reference: format!("example.test/shared@{digest}"), + tags: vec!["example.test/shared:latest".into()], + platform: "linux/amd64".into(), + }, + }); + candidate.bind_builds(&builds); + for (name, digest) in [("one", FIRST_CONTENT), ("two", SECOND_CONTENT)] { + let service = candidate + .intent() + .target + .iter() + .find(|service| service.name.as_str() == name) + .unwrap(); + assert_eq!( + service.container.image, + format!("example.test/shared@{digest}") + ); + assert_eq!(service.container.pull_policy, ployz_core::PullPolicy::Never); + } +} diff --git a/crates/ployz/tests/build_layer3/remote.rs b/crates/ployz/tests/build_layer3/remote.rs index cea7a6bb1e..f1716d5498 100644 --- a/crates/ployz/tests/build_layer3/remote.rs +++ b/crates/ployz/tests/build_layer3/remote.rs @@ -215,6 +215,7 @@ async fn request( sender .send( remote::encode(&Input::Start(Definition { + image_contexts: Default::default(), targets: vec![ployz_build::Target { name: "app".into(), platform: None, @@ -246,3 +247,152 @@ async fn terminal(response: &mut tonic::Streaming) -> Outcome { } } } + +#[tokio::test] +#[ignore = "informing: requires the privileged Ployz testkit image with Buildx"] +async fn remote_build_delivers_dependency_content_and_deploys_without_a_registry() { + let cluster = Cluster::create( + ClusterPlan::new(&format!("l3-build-803-{}", std::process::id()), 2).unwrap(), + ) + .unwrap(); + let machines = cluster.initialize_two().await.unwrap(); + let selected = machines.get(1).unwrap().id; + let destination = machines.first().unwrap().id; + let address = cluster.api_address(0).unwrap(); + let root = std::env::temp_dir().join(format!("ployz-build-803-{}", uuid::Uuid::new_v4())); + fs::create_dir_all(root.join("base")).unwrap(); + fs::create_dir_all(root.join("app")).unwrap(); + let image = format!("registry.invalid/ployz-803-{}:shared", std::process::id()); + fs::write(root.join("compose.yaml"), format!( + "name: remote\nservices:\n base:\n image: {image}\n build: ./base\n profiles: [build-only]\n app:\n image: {image}\n pull_policy: never\n x-machines: [{destination}]\n build:\n context: ./app\n additional_contexts:\n base: service:base\n" + )).unwrap(); + fs::write( + root.join("base/Dockerfile"), + "FROM alpine:3.23.3\nRUN echo dependency-output > /dependency\n", + ) + .unwrap(); + fs::write( + root.join("app/Dockerfile"), + "FROM base\nCOPY payload /payload\nCMD [\"sleep\", \"3600\"]\n", + ) + .unwrap(); + fs::write(root.join("app/payload"), "remote-application-ran\n").unwrap(); + fs::write( + root.join("docker"), + format!( + "#!/bin/sh\nprintf invoked > '{}'\nexit 99\n", + root.join("local-docker-called").display() + ), + ) + .unwrap(); + fs::set_permissions(root.join("docker"), fs::Permissions::from_mode(0o700)).unwrap(); + let output = tokio::time::timeout( + Duration::from_secs(240), + tokio::process::Command::new(env!("CARGO_BIN_EXE_ployz")) + .current_dir(&root) + .env("PATH", &root) + .env("HOME", &root) + .env("PLOYZ_CONFIG", root.join("config.yaml")) + .env("DOCKER_HOST", "unix:///no-local-docker.sock") + .args([ + "--connect", + &address, + "deploy", + &format!("--remote={selected}"), + "--yes", + "app", + ]) + .kill_on_drop(true) + .output(), + ) + .await + .unwrap() + .unwrap(); + assert!( + output.status.success(), + "{}\n{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + assert!(!root.join("local-docker-called").exists()); + let container = cluster + .machine_shell(0, "docker ps -q --filter label=ployz.service.name=app") + .unwrap(); + let container = container.trim(); + assert!(!container.is_empty(), "application was not running"); + assert_eq!( + cluster + .machine_shell( + 0, + &format!("docker exec {container} cat /dependency /payload") + ) + .unwrap(), + "dependency-output\nremote-application-ran\n" + ); + let spec_image = cluster + .machine_shell( + 0, + &format!("docker inspect {container} --format '{{{{.Config.Image}}}}'"), + ) + .unwrap(); + assert!(spec_image.contains("@sha256:"), "{spec_image}"); + assert_eq!( + cluster + .machine_shell(1, "docker ps -q --filter label=ployz.service.name=app") + .unwrap(), + "" + ); + // The same captured graph also works when each Build runs on a different Machine. + let load = ployz::compose::LoadOptions { + command: "build".into(), + all_profiles: true, + working_dir: Some(root.clone()), + ..Default::default() + }; + let mut project = ployz::compose::load_project(&load).unwrap(); + let options = ployz::compose::BuildOptions { + services: vec!["app".into()], + ..Default::default() + }; + let plan = ployz::compose::plan_build(&project, &options).unwrap(); + let captured = ployz::compose::capture_build(&plan, &options, &mut project).unwrap(); + let mut client = ployz::connect::connect(&root.join("config.yaml"), Some(&address), None) + .await + .unwrap(); + let images = captured + .execute_on_machines( + &client, + &std::collections::BTreeMap::from([ + ("base".into(), selected), + ("app".into(), destination), + ]), + CancellationToken::new(), + |_| {}, + ) + .await + .unwrap(); + let app = images.iter().find(|image| image.name == "app").unwrap(); + let result = ployz::image::push_from_machine( + &mut client, + &app.built, + destination, + &[selected.to_string()], + ) + .await + .unwrap(); + assert_eq!(result.successes.len(), 1); + assert!(result.failures.is_empty() && result.omissions.is_empty()); + assert_eq!( + cluster + .machine_shell( + 1, + &format!( + "docker run --rm --pull=never {} cat /dependency /payload", + app.built.reference, + ) + ) + .unwrap(), + "dependency-output\nremote-application-ran\n" + ); + fs::remove_dir_all(root).unwrap(); +} diff --git a/crates/ployz/tests/deploy_client/support.rs b/crates/ployz/tests/deploy_client/support.rs index bb99b6c950..0033ca5de4 100644 --- a/crates/ployz/tests/deploy_client/support.rs +++ b/crates/ployz/tests/deploy_client/support.rs @@ -34,10 +34,15 @@ use tonic::{Request, Response, Status, Streaming, transport::Server}; #[path = "../support/inspect_telemetry.rs"] mod inspect_telemetry_fixture; +#[path = "../support/build.rs"] +mod build_fixture; +pub(super) use build_fixture::BuildFixture; + pub(super) type ObservationRequest = (MachineId, Vec, u64); #[derive(Clone)] pub(super) struct DeployService { + pub(super) builds: Option>, machines: Vec, create_volume_error: Option, create_volume_verification_error: Option, @@ -62,6 +67,7 @@ pub(super) struct DeployService { impl DeployService { pub(super) fn new(machine: MachineObservation) -> Self { Self { + builds: None, machines: vec![machine], create_volume_error: None, create_volume_verification_error: None, @@ -86,6 +92,7 @@ impl DeployService { pub(super) fn empty() -> Self { Self { + builds: None, machines: Vec::new(), create_volume_error: None, create_volume_verification_error: None, @@ -239,13 +246,16 @@ impl MachineRpc for DeployService { } type ExecStream = tokio_stream::wrappers::ReceiverStream>; - type BuildStream = tokio_stream::Empty>; + type BuildStream = tokio_stream::wrappers::ReceiverStream>; async fn build( &self, - _request: Request>, + request: Request>, ) -> Result, Status> { - Err(Status::unimplemented("Build is not used by this fixture")) + self.builds + .clone() + .ok_or_else(|| Status::unimplemented("Build is not used by this fixture"))? + .stream(request) } type ContainerLogsStream = tokio_stream::Empty>; type MachineLogsStream = tokio_stream::Empty>; @@ -253,7 +263,7 @@ impl MachineRpc for DeployService { async fn describe_contract( &self, - _request: Request, + request: Request, ) -> Result, Status> { let mut capabilities: BTreeSet<_> = [CapabilityName::parse(MACHINE_STORAGE_OBSERVATION_CAPABILITY).unwrap()].into(); @@ -261,11 +271,20 @@ impl MachineRpc for DeployService { capabilities .insert(CapabilityName::parse(GET_CONTAINER_OBSERVATIONS_CAPABILITY).unwrap()); } + if self.builds.is_some() { + capabilities.extend( + [ + ployz_core::BUILD_CAPABILITY, + ployz_core::ENSURE_IMAGE_INGEST_CAPABILITY, + ployz_core::PULL_IMAGE_FROM_MACHINE_CAPABILITY, + ] + .map(|name| CapabilityName::parse(name).unwrap()), + ); + } encoded(RpcResponse::from(ContractDescription { - machine_id: self - .machines - .first() - .map(|machine| machine.machine.id) + machine_id: machine_from_metadata(&request) + .ok() + .or_else(|| self.machines.first().map(|machine| machine.machine.id)) .unwrap_or_else(MachineId::random), protocol_major: PROTOCOL_MAJOR, daemon_version: "test".into(), @@ -709,24 +728,48 @@ impl MachineRpc for DeployService { } async fn list_images( &self, - _request: Request, + request: Request, ) -> Result, Status> { - encoded(RpcResponse::from(MachineImages { - containerd_store: false, - images: Vec::new(), - })) + let machine = machine_from_metadata(&request)?; + let images = self + .builds + .as_ref() + .map(|builds| builds.images(machine)) + .unwrap_or(MachineImages { + containerd_store: self.builds.is_some(), + images: Vec::new(), + }); + encoded(RpcResponse::from(images)) } async fn ensure_image_ingest( &self, - _request: Request, + request: Request, ) -> Result, Status> { + if let Some(builds) = &self.builds { + let machine = machine_from_metadata(&request)?; + let machine = self + .machines + .iter() + .find(|observed| observed.machine.id == machine) + .unwrap(); + return encoded(RpcResponse::from(builds.open(&machine.machine))); + } self.record_mutation(); unused() } async fn pull_image_from_machine( &self, - _request: Request, + request: Request, ) -> Result, Status> { + if let Some(builds) = &self.builds { + let machine = machine_from_metadata(&request)?; + let RpcRequestBody::PullImageFromMachine(pull) = + request.into_inner().decode_request().unwrap().body + else { + panic!("expected peer pull"); + }; + return encoded(builds.pull(machine, pull)); + } self.record_mutation(); unused() } diff --git a/crates/ployz/tests/support/build.rs b/crates/ployz/tests/support/build.rs new file mode 100644 index 0000000000..f08a01f4dd --- /dev/null +++ b/crates/ployz/tests/support/build.rs @@ -0,0 +1,152 @@ +//! Owned transport stand-in. It records image identities and RPCs, never proves +//! that Docker built or ran an image; build_layer3 supplies that evidence. + +use ployz_build::{ + Output, + remote::{self, Event, Input, Outcome}, +}; +use ployz_core::{MachineId, MachineImages, OpaquePayload, PullImageFromMachineRequest}; +use std::{ + collections::BTreeMap, + sync::{Arc, Mutex}, +}; +use tokio::sync::mpsc; +use tokio_stream::wrappers::ReceiverStream; +use tonic::{Request, Response, Status}; + +#[derive(Default)] +pub struct BuildFixture { + pub terminal: Mutex>, + pub definitions: Mutex>, + pub stores: Mutex>, + pub opened: Mutex>, + pub pulls: Mutex>, + pub pull_failures: Mutex>, +} + +impl BuildFixture { + pub fn images(&self, machine: MachineId) -> MachineImages { + self.stores + .lock() + .unwrap() + .get(&machine) + .cloned() + .unwrap_or(MachineImages { + containerd_store: true, + images: Vec::new(), + }) + } + + pub fn open(&self, machine: &ployz_core::Machine) -> ployz_core::ImageIngestOpened { + self.opened.lock().unwrap().push(machine.id); + ployz_core::ImageIngestOpened { + destination: ployz_core::ImageIngestDestination { + management_address: machine.management_address(), + port: ployz_core::UNREGISTRY_PORT, + }, + } + } + + pub fn pull( + &self, + machine: MachineId, + pull: PullImageFromMachineRequest, + ) -> ployz_core::RpcResponse { + self.pulls.lock().unwrap().push((machine, pull)); + self.pull_failures + .lock() + .unwrap() + .get(&machine) + .cloned() + .map_or_else( + || ployz_core::RpcResponse::from(ployz_core::ImagePulled {}), + ployz_core::RpcResponse::from, + ) + } + + #[expect(clippy::result_large_err)] // tonic fixes the public RPC error type. + pub fn stream( + self: Arc, + request: Request>, + ) -> Result>>, Status> { + let machine_id = + MachineId::parse(request.metadata().get("machine").unwrap().to_str().unwrap()) + .map_err(|error| Status::invalid_argument(error.to_string()))?; + let (sender, receiver) = mpsc::channel(2); + tokio::spawn(async move { + let mut request = request.into_inner(); + let Input::Start(definition) = + remote::decode(&request.message().await.unwrap().unwrap()).unwrap() + else { + panic!("expected Build definition"); + }; + let offset = { + let mut definitions = self.definitions.lock().unwrap(); + let offset = definitions + .iter() + .map(|definition| definition.targets.len()) + .sum::(); + definitions.push(definition.clone()); + offset + }; + sender + .send(Ok(remote::encode(&Event::Admitted { machine_id }).unwrap())) + .await + .unwrap(); + let mut upload = remote::Upload::new().unwrap(); + loop { + let payload = request.message().await.unwrap().unwrap(); + let frame: Input = remote::decode(&payload).unwrap(); + let finish = matches!(frame, Input::Finish); + upload.accept(frame).unwrap(); + if finish { + break; + } + } + let terminal = self.terminal.lock().unwrap().clone(); + let outcome = terminal.unwrap_or_else(|| match definition.output { + Output::Validate => Outcome::Validated { machine_id }, + Output::Registry => Outcome::Published { machine_id }, + Output::Load => { + let images = definition + .targets + .iter() + .enumerate() + .map(|(index, _)| ployz_build::BuiltImage { + reference: format!( + "registry.invalid/shared@sha256:{}", + (offset + index + 1).to_string().repeat(64) + ), + tags: vec!["registry.invalid/shared:latest".into()], + platform: "linux/amd64".into(), + }) + .collect::>(); + let mut stores = self.stores.lock().unwrap(); + let store = stores.entry(machine_id).or_insert(MachineImages { + containerd_store: true, + images: Vec::new(), + }); + // Keep content after another Build overwrites the requested tag. + for stored in &mut store.images { + stored.repo_tags.clear(); + } + store + .images + .extend(images.iter().map(|image| ployz_core::ImageSummary { + id: image.reference.split_once('@').unwrap().1.into(), + repo_tags: image.tags.clone(), + created: 0, + size: 1, + containers: 0, + platforms: vec![image.platform.clone()], + })); + Outcome::Images { machine_id, images } + } + }); + let _ = sender + .send(Ok(remote::encode(&Event::Finished(outcome)).unwrap())) + .await; + }); + Ok(Response::new(ReceiverStream::new(receiver))) + } +} diff --git a/crates/ployzd/src/build.rs b/crates/ployzd/src/build.rs index b9d104be6a..3f1b0afbe0 100644 --- a/crates/ployzd/src/build.rs +++ b/crates/ployzd/src/build.rs @@ -53,7 +53,10 @@ async fn attempt( }, _ => return failed(Stage::Admission, "Build request ended before admission"), }; - if definition.targets.is_empty() || definition.targets.len() > 128 { + if definition.targets.is_empty() + || definition.targets.len() > 128 + || definition.image_contexts.len() > 128 + { return failed( Stage::Admission, "Build must name between one and 128 targets", @@ -151,7 +154,7 @@ async fn pump( fn receive_and_execute( machine_id: MachineId, - definition: Definition, + mut definition: Definition, mut source: mpsc::Receiver, admission: Admission, policy: HostPolicy, @@ -214,6 +217,21 @@ fn receive_and_execute( } }; progress(Progress::Stage(Stage::Upload)); + // Keep bytes on the Machines. Reuse the Docker peer-pull bridge because + // Buildx cannot reliably encode IPv6 registry keys in its TOML config. + let mut proxies = Vec::new(); + for context in definition.image_contexts.values_mut() { + let proxy = match tokio::runtime::Handle::current() + .block_on(crate::docker::ImageProxy::open(context.source)) + { + Ok(proxy) => proxy, + Err(error) => return failed(Stage::Preparation, error.to_string()), + }; + context.source.management_address = + ployz_core::ManagementAddress(std::net::Ipv4Addr::LOCALHOST.to_ipv6_mapped()); + context.source.port = proxy.port; + proxies.push(proxy); + } match upload.execute(&definition, admission, Some(&policy.docker), &progress) { Ok(images) => match definition.output { Output::Load => Outcome::Images { machine_id, images }, diff --git a/crates/ployzd/src/build/tests.rs b/crates/ployzd/src/build/tests.rs index 5074713ed8..9d8f013686 100644 --- a/crates/ployzd/src/build/tests.rs +++ b/crates/ployzd/src/build/tests.rs @@ -88,6 +88,7 @@ impl Fixture { sender .send( remote::encode(&Input::Start(Definition { + image_contexts: Default::default(), targets: vec![ployz_build::Target { name: "api".into(), platform: None, @@ -368,6 +369,7 @@ async fn upload_timeout_stops_before_execution_and_releases_admission() { let (sender, receiver) = mpsc::channel(2); sender .send(Ok(remote::encode(&Input::Start(Definition { + image_contexts: Default::default(), targets: vec![ployz_build::Target { name: "api".into(), platform: None, @@ -453,7 +455,12 @@ async fn terminal_failures_preserve_completed_images_and_uncertain_targets() { if failure != "fail-cleanup" { assert_eq!( work.0.get("web"), - Some(&ployz_build::TargetEvidence::Unknown) + Some(&if failure == "fail-after-output" { + // The next target is never submitted after this step fails. + ployz_build::TargetEvidence::Unattempted + } else { + ployz_build::TargetEvidence::Unknown + }) ); } } diff --git a/crates/ployzd/src/docker/mod.rs b/crates/ployzd/src/docker/mod.rs index d5c7ac769b..05ed7ebf89 100644 --- a/crates/ployzd/src/docker/mod.rs +++ b/crates/ployzd/src/docker/mod.rs @@ -4,6 +4,7 @@ mod lifecycle; mod managed_service; mod observe; mod peer_pull; +pub(crate) use peer_pull::ImageProxy; mod spec_store; mod stream; mod telemetry; @@ -611,6 +612,8 @@ fn container_address(inspected: &RawContainerInspect) -> Option RpcErrorCode::Conflict, - Self::ProvisionedStorageUnsupported => RpcErrorCode::Unsupported, + Self::ProvisionedStorageUnsupported | Self::UnsupportedImageStore => RpcErrorCode::Unsupported, Self::VolumeCreatedButUnverified { .. } | Self::StorageUnobservable | Self::EventStreamClosed diff --git a/crates/ployzd/src/docker/peer_pull.rs b/crates/ployzd/src/docker/peer_pull.rs index 626bc007eb..12a703d4ff 100644 --- a/crates/ployzd/src/docker/peer_pull.rs +++ b/crates/ployzd/src/docker/peer_pull.rs @@ -23,53 +23,99 @@ pub(crate) async fn pull_from_ingest( image: &str, source: ImageIngestDestination, ) -> Result<(), Error> { - let listener = TcpListener::bind("127.0.0.1:0") - .await - .map_err(|error| Error::PeerPull(error.to_string()))?; - let port = listener - .local_addr() - .map_err(|error| Error::PeerPull(error.to_string()))? - .port(); - let source_addr = SocketAddr::from((source.management_address.0, source.port)); - let proxy = tokio::spawn(proxy_to_source(listener, source_addr)); - let pulled = localhost_registry_reference(port, image); - let result = async { - docker_cli(["pull", &pulled]).await?; - docker_cli(["tag", &pulled, image]).await?; - let _ = docker_cli(["image", "rm", &pulled]).await; - Ok(()) + let retained = if image.contains('@') { + ployz_build::remote::validate_remote_context(&format!("docker-image://{image}")) + .map_err(|error| Error::PeerPull(error.to_string()))?; + if !super::LocalDocker::connect()? + .uses_containerd_store() + .await? + { + return Err(Error::UnsupportedImageStore); + } + Some(image.replace("@sha256:", ":ployz-sha256-")) + } else { + None + }; + let proxy = ImageProxy::open(source).await?; + let pulled = localhost_registry_reference(proxy.port, image); + docker_cli(["pull", &pulled]).await?; + // Docker cannot tag repository@digest. A content-specific tag retains + // that repository's digest without another client's tag overwriting it. + docker_cli(["tag", &pulled, retained.as_deref().unwrap_or(image)]).await?; + if let Some((_, digest)) = image.split_once('@') { + let descriptor = docker_cli([ + "image", + "inspect", + image, + "--format", + "{{json .Descriptor}}", + ]) + .await?; + let descriptor: serde_json::Value = serde_json::from_str(&descriptor)?; + if descriptor.get("digest").and_then(serde_json::Value::as_str) != Some(digest) { + return Err(Error::PeerPull(format!( + "the image store did not retain {image}" + ))); + } } - .await; - proxy.abort(); - result + let _ = docker_cli(["image", "rm", &pulled]).await; + Ok(()) } fn localhost_registry_reference(port: u16, image: &str) -> String { format!("127.0.0.1:{port}/{image}") } -async fn proxy_to_source(listener: TcpListener, source: SocketAddr) { - loop { - let Ok((mut inbound, _)) = listener.accept().await else { - break; - }; - tokio::spawn(async move { - let Ok(mut outbound) = TcpStream::connect(source).await else { - return; - }; - let _ = copy_bidirectional(&mut inbound, &mut outbound).await; +/// A connection-scoped loopback bridge to a Machine image server. Both Docker +/// and BuildKit accept loopback HTTP without changing global registry policy. +pub(crate) struct ImageProxy { + pub(crate) port: u16, + task: tokio::task::JoinHandle<()>, +} + +impl ImageProxy { + pub(crate) async fn open(source: ImageIngestDestination) -> Result { + let listener = TcpListener::bind("127.0.0.1:0") + .await + .map_err(|error| Error::PeerPull(error.to_string()))?; + let port = listener + .local_addr() + .map_err(|error| Error::PeerPull(error.to_string()))? + .port(); + let source = SocketAddr::from((source.management_address.0, source.port)); + let task = tokio::spawn(async move { + let mut connections = tokio::task::JoinSet::new(); + loop { + tokio::select! { + accepted = listener.accept() => { + let Ok((mut inbound, _)) = accepted else { break }; + connections.spawn(async move { + if let Ok(mut outbound) = TcpStream::connect(source).await { + let _ = copy_bidirectional(&mut inbound, &mut outbound).await; + } + }); + } + _ = connections.join_next(), if !connections.is_empty() => {} + } + } }); + Ok(Self { port, task }) + } +} +impl Drop for ImageProxy { + fn drop(&mut self) { + self.task.abort(); } } -async fn docker_cli(args: [&str; N]) -> Result<(), Error> { +async fn docker_cli(args: [&str; N]) -> Result { let output = Command::new("docker") .args(args) .output() .await .map_err(|error| Error::PeerPull(error.to_string()))?; if output.status.success() { - Ok(()) + Ok(String::from_utf8_lossy(&output.stdout).into_owned()) } else { Err(Error::PeerPull( String::from_utf8_lossy(&output.stderr).trim().into(), From e09f5676bad36ac5afcaa5e7aec17bf6a156b178 Mon Sep 17 00:00:00 2001 From: Codex Date: Wed, 9 Sep 2026 03:59:10 +0000 Subject: [PATCH 2/3] Release temporary build tags and validate destination architectures --- crates/ployz-build/src/lib.rs | 2 + crates/ployz-build/src/received_recipe.rs | 18 ++ crates/ployz-build/src/remote.rs | 3 + crates/ployz-build/src/retention.rs | 126 +++++++++ crates/ployz-build/src/upload.rs | 13 +- crates/ployz/src/compose/build.rs | 107 +++----- crates/ployz/src/compose/capture.rs | 53 ++++ crates/ployz/src/compose/remote_build.rs | 253 ++++++++++-------- crates/ployz/src/compose/remote_steps.rs | 82 +++++- crates/ployz/src/deploy/pipeline.rs | 7 +- .../ployz/src/handlers/deploy_remote_tests.rs | 64 ++++- crates/ployz/src/image.rs | 2 +- crates/ployz/src/image/built.rs | 65 +++++ crates/ployz/tests/build.rs | 43 +-- crates/ployz/tests/build/capture.rs | 44 +++ crates/ployz/tests/build_layer3/remote.rs | 24 ++ crates/ployz/tests/support/build.rs | 8 +- crates/ployzd/Cargo.toml | 2 +- crates/ployzd/src/build.rs | 64 +++-- crates/ployzd/src/build/tests.rs | 37 +++ crates/ployzd/src/machine/mod.rs | 6 +- 21 files changed, 767 insertions(+), 256 deletions(-) create mode 100644 crates/ployz-build/src/retention.rs diff --git a/crates/ployz-build/src/lib.rs b/crates/ployz-build/src/lib.rs index 3f070b76f7..c4d3eb858b 100644 --- a/crates/ployz-build/src/lib.rs +++ b/crates/ployz-build/src/lib.rs @@ -16,6 +16,8 @@ mod image_contexts; mod policy; pub use policy::clear_cache; mod received_recipe; +mod retention; +pub use retention::{ImageRetention, RetainedImages}; pub mod remote; mod upload; diff --git a/crates/ployz-build/src/received_recipe.rs b/crates/ployz-build/src/received_recipe.rs index 11dfa4004d..1bd406d9fc 100644 --- a/crates/ployz-build/src/received_recipe.rs +++ b/crates/ployz-build/src/received_recipe.rs @@ -56,6 +56,11 @@ pub fn validate_capture( return Err("invalid completed Build image context".into()); } } + if definition.retained_tags.len() > names.len() + || (!definition.retained_tags.is_empty() && definition.output != crate::Output::Load) + { + return Err("invalid temporary Build tags".into()); + } let context_names = names .iter() .copied() @@ -71,6 +76,17 @@ pub fn validate_capture( if services.len() != names.len() { return Err("Build recipe targets differ from the admitted request".into()); } + for tag in &definition.retained_tags { + if !services.values().any(|service| { + service + .get("build") + .and_then(|build| build.get("tags")) + .and_then(Value::as_sequence) + .is_some_and(|tags| tags.iter().any(|value| value.as_str() == Some(tag))) + }) { + return Err("temporary Build tag is absent from the captured recipe".into()); + } + } for (name, service) in services { let name = text(name)?; if !names.contains(name) { @@ -472,6 +488,7 @@ mod tests { ) .unwrap(); let definition = Definition { + retained_tags: Vec::new(), image_contexts: Default::default(), targets: vec![crate::Target { name: "api".into(), @@ -518,6 +535,7 @@ mod tests { fs::write(root.join("private/key"), "captured-key").unwrap(); fs::write(root.join("source/key"), "source-content").unwrap(); let definition = Definition { + retained_tags: Vec::new(), image_contexts: Default::default(), targets: vec![crate::Target { name: "api".into(), diff --git a/crates/ployz-build/src/remote.rs b/crates/ployz-build/src/remote.rs index e4552d2247..5b7956107a 100644 --- a/crates/ployz-build/src/remote.rs +++ b/crates/ployz-build/src/remote.rs @@ -48,6 +48,9 @@ pub const CHUNK_SIZE: usize = 32 * 1024; #[derive(Clone, Debug, Serialize, Deserialize)] #[serde(deny_unknown_fields)] pub struct Definition { + /// Command-generated temporary tags, retained until the response stream closes. + #[serde(default)] + pub retained_tags: Vec, /// Completed Service contexts, bound to immutable content and serving endpoints. #[serde(default)] pub image_contexts: std::collections::BTreeMap, diff --git a/crates/ployz-build/src/retention.rs b/crates/ployz-build/src/retention.rs new file mode 100644 index 0000000000..034d00de5f --- /dev/null +++ b/crates/ployz-build/src/retention.rs @@ -0,0 +1,126 @@ +//! Command-owned temporary tags, independent of captured source and credentials. + +use crate::{Deadline, Docker, Streams}; +use std::{ + collections::BTreeMap, + fs, + os::unix::fs::DirBuilderExt as _, + path::{Path, PathBuf}, + time::Duration, +}; + +/// Removes only this command's temporary tags when its image results are released. +pub struct ImageRetention { + tags: Vec, + program: PathBuf, + environment: BTreeMap, + root: PathBuf, +} + +/// Completed host images and the guard that keeps their temporary tags alive. +pub struct RetainedImages { + /// Immutable content reported by the completed Build. + pub images: Vec, + /// Release after dependent builds and image delivery finish. + pub retention: ImageRetention, +} + +impl ImageRetention { + /// Own cleanup of the explicitly generated tags, never user image references. + /// # Errors + /// Rejects references outside the reserved UUID-tag namespace or unavailable private staging. + pub fn new( + tags: Vec, + docker: Option<&Path>, + mut environment: BTreeMap, + ) -> Result { + for tag in &tags { + let valid = tag + .parse::() + .ok() + .is_some_and(|reference| { + reference.digest().is_none() + && reference + .tag() + .and_then(|tag| tag.strip_prefix("ployz-build-")) + .is_some_and(|id| uuid::Uuid::parse_str(id).is_ok()) + }); + if !valid { + return Err(crate::BuildError::Request( + "invalid temporary Build tag".into(), + )); + } + } + // Image removal needs the captured local socket, not registry credentials + // or the source tree. Docker uses an empty, private default configuration. + let root = + std::env::temp_dir().join(format!("ployz-image-retention-{}", uuid::Uuid::new_v4())); + fs::DirBuilder::new() + .mode(0o700) + .create(&root) + .map_err(|error| { + crate::BuildError::Request(format!("create image retention directory: {error}")) + })?; + environment + .retain(|name, _| matches!(name.as_str(), "PATH" | "DOCKER_HOST" | "DOCKER_CONTEXT")); + environment.insert("HOME".into(), root.to_string_lossy().into_owned()); + environment.insert("DOCKER_CONFIG".into(), root.to_string_lossy().into_owned()); + Ok(Self { + tags, + program: docker.unwrap_or_else(|| Path::new("docker")).to_owned(), + environment, + root, + }) + } +} + +impl Drop for ImageRetention { + fn drop(&mut self) { + if self.tags.is_empty() { + let _ = fs::remove_dir_all(&self.root); + return; + } + let docker = Docker { + program: &self.program, + environment: &self.environment, + working_dir: &self.root, + deadline: Deadline::starting_now(Duration::from_secs(10)), + cancellation: None, + progress: None, + }; + let arguments: Vec<_> = ["image", "rm"] + .into_iter() + .chain(self.tags.iter().map(String::as_str)) + .collect(); + if let Err(error) = docker.require_local().and_then(|()| { + docker.run( + "release temporary Build tags", + &arguments, + Streams::Captured, + ) + }) { + eprintln!("Could not release temporary Build tags: {error}"); + } + let _ = fs::remove_dir_all(&self.root); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn retention_cannot_delete_user_tags_or_digest_references() { + for reference in [ + "example.test/api:latest", + "example.test/api:ployz-build-not-a-uuid", + "--all", + "example.test/api@sha256:1111111111111111111111111111111111111111111111111111111111111111", + ] { + assert!( + ImageRetention::new(vec![reference.into()], None, BTreeMap::new()).is_err(), + "{reference}" + ); + } + } +} diff --git a/crates/ployz-build/src/upload.rs b/crates/ployz-build/src/upload.rs index ba5af469d3..02c278021a 100644 --- a/crates/ployz-build/src/upload.rs +++ b/crates/ployz-build/src/upload.rs @@ -2,7 +2,7 @@ use crate::remote::InputError; use crate::{ - Admission, BuildError, BuiltImage, Progress, Request, + Admission, BuildError, Progress, Request, remote::{CHUNK_SIZE, Definition, Input, Kind}, }; use std::{ @@ -217,7 +217,7 @@ impl Upload { } impl CompletedUpload { - /// Consume only a completed capture. The remote host supplies its own PATH + /// Execute only a completed capture. The remote host supplies its own PATH /// and Docker socket; client environment and plugin paths are never used. /// # Errors /// Returns recipe validation or host execution failure. @@ -227,7 +227,7 @@ impl CompletedUpload { admission: Admission, docker: Option<&Path>, progress: &(dyn Fn(Progress) + Sync), - ) -> Result, BuildError> { + ) -> Result { let mut upload = self.0; let railpack = crate::received_recipe::validate_capture(&upload.root, definition) .map_err(|error| BuildError::Request(error.to_string()))?; @@ -253,6 +253,11 @@ impl CompletedUpload { .into_owned(), ), ]); + let retention = crate::ImageRetention::new( + definition.retained_tags.clone(), + docker, + environment.clone(), + )?; let result = crate::execute_admitted( &Request { image_contexts: &definition.image_contexts, @@ -273,7 +278,7 @@ impl CompletedUpload { // Unconfirmed processes may still read private attempt files. Keep them // protected alongside quarantined builder ownership until safe cleanup. upload.retain = result.as_ref().is_err_and(|error| error.is_unknown()); - result + result.map(|images| crate::RetainedImages { images, retention }) } } diff --git a/crates/ployz/src/compose/build.rs b/crates/ployz/src/compose/build.rs index abb02618e7..bd940ddcf1 100644 --- a/crates/ployz/src/compose/build.rs +++ b/crates/ployz/src/compose/build.rs @@ -1,6 +1,7 @@ use std::{ collections::{BTreeMap, BTreeSet}, path::Path, + sync::{Arc, Mutex}, }; use ployz_build::{BuiltImage, Output}; @@ -70,6 +71,7 @@ pub struct CapturedBuild { options: BuildOptions, environment: BTreeMap, inputs: BuildInputs, + retained_tags: BTreeMap, } /// Where a completed image is available; Machine identity is already resolved. @@ -83,7 +85,7 @@ pub enum BuildLocation { } /// A Service whose image this command built, bound to the content produced. -#[derive(Clone, Debug, PartialEq)] +#[derive(Clone, Debug)] pub struct BuiltService { /// Raw Compose Build target name, including build-only dependencies. /// Build names may contain dots/underscores, unlike Deploy Service names. @@ -96,6 +98,32 @@ pub struct BuiltService { pub machines: Vec, /// The image this command built for it. pub built: BuiltImage, + pub(super) _retention: Option, +} + +#[derive(Clone)] +pub(super) enum BuildRetention { + Local { + _tags: Arc, + }, + Remote { + _stream: Arc>>, + }, +} +impl std::fmt::Debug for BuildRetention { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str("BuildRetention") + } +} + +impl PartialEq for BuiltService { + fn eq(&self, other: &Self) -> bool { + self.name == other.name + && self.location == other.location + && self.image == other.image + && self.machines == other.machines + && self.built == other.built + } } impl BuiltService { @@ -158,6 +186,7 @@ pub fn capture_build( let mut plan = plan.to_vec(); let mut secret_names = BTreeSet::new(); let mut targets = Vec::new(); + let mut retained_tags = BTreeMap::new(); let mut railpack_recipes = Vec::new(); for service in &mut plan { let image = service.image.clone(); @@ -202,6 +231,7 @@ pub fn capture_build( reference.repository(), uuid::Uuid::new_v4() ); + retained_tags.insert(service.name.clone(), retained.clone()); let tags = build .entry(Value::String("tags".into())) .or_insert_with(|| Value::Sequence(vec![Value::String(service.image.clone())])); @@ -339,6 +369,7 @@ pub fn capture_build( .map(|(key, value)| (key.clone(), value.clone())) .collect(), inputs, + retained_tags, }) } @@ -554,69 +585,6 @@ fn effective_build_args( } impl CapturedBuild { - /// Execute this capture on one resolved Machine over the authenticated - /// Ployz stream. No local Docker or local image store is consulted. - pub async fn execute_remote( - self, - client: &crate::connect::Client, - machine_id: ployz_core::MachineId, - cancellation: tokio_util::sync::CancellationToken, - progress: impl Fn(ployz_build::Progress), - ) -> ployz_build::remote::Outcome { - if self.options.output == Output::Load { - let locations = self - .targets - .iter() - .map(|target| (target.name.clone(), machine_id)) - .collect(); - return match self - .execute_remote_steps(client, &locations, cancellation, progress) - .await - { - Ok(images) => ployz_build::remote::Outcome::Images { - machine_id, - images: images.into_iter().map(|service| service.built).collect(), - }, - Err(outcome) => outcome, - }; - } - let definition = ployz_build::remote::Definition { - image_contexts: Default::default(), - targets: self.targets, - output: self.options.output, - no_cache: self.options.no_cache, - pull: self.options.pull, - }; - super::remote_build::execute( - self.inputs, - definition, - client, - machine_id, - cancellation, - progress, - ) - .await - } - - /// Execute remotely and bind each completed output to its captured Service. - /// # Errors - /// Failed, unknown, or image-less outcomes cannot be used by Deploy. - pub async fn execute_remote_images( - self, - client: &crate::connect::Client, - machine_id: ployz_core::MachineId, - cancellation: tokio_util::sync::CancellationToken, - progress: impl Fn(ployz_build::Progress), - ) -> Result, ComposeError> { - let locations = self - .targets - .iter() - .map(|target| (target.name.clone(), machine_id)) - .collect(); - self.execute_on_machines(client, &locations, cancellation, progress) - .await - } - /// Build this capture through the shared runner, without reading the /// original sources again. /// @@ -641,6 +609,16 @@ impl CapturedBuild { .to_string_lossy() .into_owned(), ); + let retention = BuildRetention::Local { + _tags: Arc::new( + ployz_build::ImageRetention::new( + self.retained_tags.values().cloned().collect(), + docker, + environment.clone(), + ) + .map_err(|error| invalid_build(&error.to_string()))?, + ), + }; let images = ployz_build::execute(&ployz_build::Request { image_contexts: &BTreeMap::new(), compose_file: Path::new("compose.yaml"), @@ -675,6 +653,7 @@ impl CapturedBuild { image: service.image.clone(), machines: service.machines.clone(), built, + _retention: Some(retention.clone()), }) .collect()) } diff --git a/crates/ployz/src/compose/capture.rs b/crates/ployz/src/compose/capture.rs index 7a00fec185..136b6fb74f 100644 --- a/crates/ployz/src/compose/capture.rs +++ b/crates/ployz/src/compose/capture.rs @@ -100,3 +100,56 @@ impl ComposeProject { } } } + +#[cfg(test)] +#[test] +fn deploy_binds_each_service_to_its_build_when_requested_tags_are_shared() { + use super::build::BuiltService; + use crate::compose::{BuildLocation, parse_normalized}; + let first_content = format!("sha256:{}", "1".repeat(64)); + let second_content = format!("sha256:{}", "2".repeat(64)); + let project = parse_normalized( + "services: {one: {image: 'example.test/shared:latest', build: .}, two: {image: 'example.test/shared:latest', build: .}}", + ".", + ).unwrap(); + let mut candidate = project.capture( + ployz_core::ProjectName::parse("app").unwrap(), + Default::default(), + vec![], + None, + vec![], + ); + let builds = [ + ("one", first_content.as_str()), + ("two", second_content.as_str()), + ] + .map(|(name, digest)| BuiltService { + name: name.into(), + _retention: None, + image: "example.test/shared:latest".into(), + machines: vec![], + location: BuildLocation::Machine(ployz_core::MachineId::parse("a".repeat(32)).unwrap()), + built: ployz_build::BuiltImage { + reference: format!("example.test/shared@{digest}"), + tags: vec!["example.test/shared:latest".into()], + platform: "linux/amd64".into(), + }, + }); + candidate.bind_builds(&builds); + for (name, digest) in [ + ("one", first_content.as_str()), + ("two", second_content.as_str()), + ] { + let service = candidate + .intent() + .target + .iter() + .find(|service| service.name.as_str() == name) + .unwrap(); + assert_eq!( + service.container.image, + format!("example.test/shared@{digest}") + ); + assert_eq!(service.container.pull_policy, ployz_core::PullPolicy::Never); + } +} diff --git a/crates/ployz/src/compose/remote_build.rs b/crates/ployz/src/compose/remote_build.rs index d91963cc61..87664c7cde 100644 --- a/crates/ployz/src/compose/remote_build.rs +++ b/crates/ployz/src/compose/remote_build.rs @@ -8,11 +8,34 @@ use ployz_build::{ remote::{self, Definition, Event, Input, Outcome}, }; use ployz_core::{MachineId, MachineTarget}; -use std::time::Duration; +use std::{ + sync::{Arc, Mutex}, + time::Duration, +}; use tokio::sync::mpsc; use tokio_stream::wrappers::ReceiverStream; use tokio_util::sync::CancellationToken; +pub(super) enum Completion { + Images { + machine_id: MachineId, + images: Vec, + stream: Arc>>, + }, + Report(Outcome), +} +impl Completion { + /// Explicitly release temporary image retention when only a terminal report is needed. + pub(super) fn into_outcome(self) -> Outcome { + match self { + Self::Images { + machine_id, images, .. + } => Outcome::Images { machine_id, images }, + Self::Report(outcome) => outcome, + } + } +} + pub(super) async fn execute( inputs: BuildInputs, definition: Definition, @@ -20,129 +43,149 @@ pub(super) async fn execute( machine_id: MachineId, cancellation: CancellationToken, progress: impl Fn(Progress), -) -> Outcome { - let mut evidence = ployz_build::WorkEvidence::new(&definition.targets); - if cancellation.is_cancelled() { - return failed(Stage::Admission, "Build cancelled before submission").with_work(evidence); - } - if let Err(error) = remote::validate_capture(inputs.root(), &definition) { - return failed(Stage::Preparation, error.to_string()).with_work(evidence); - } - let expected = definition.targets.len(); - let output = definition.output; - let (sender, receiver) = mpsc::channel(2); - let start = match remote::encode(&Input::Start(definition)) { - Ok(start) => start, - Err(message) => return failed(Stage::Admission, message.to_string()), - }; - sender.send(start).await.expect("request receiver is owned"); - let target = MachineTarget::from(&machine_id); - let mut responses = match tokio::time::timeout( - Duration::from_secs(10), - client.build_stream(&target, ReceiverStream::new(receiver)), - ) - .await - { - Ok(Ok(responses)) => responses, - Ok(Err(error)) => { - return failed( - Stage::Admission, - format!("Machine {machine_id} cannot accept the Build: {error}"), - ); +) -> Completion { + let mut retained = None; + let outcome = async { + let mut evidence = ployz_build::WorkEvidence::new(&definition.targets); + if cancellation.is_cancelled() { + return failed(Stage::Admission, "Build cancelled before submission").with_work(evidence); } - Err(_) => { - return failed( - Stage::Admission, - format!("Machine {machine_id} did not answer; source was not uploaded"), - ); + if let Err(error) = remote::validate_capture(inputs.root(), &definition) { + return failed(Stage::Preparation, error.to_string()).with_work(evidence); } - }; - match tokio::time::timeout(Duration::from_secs(10), responses.message()).await { - Ok(Ok(Some(payload))) => match remote::decode::(&payload) { - Ok(Event::Admitted { - machine_id: admitted, - }) if admitted == machine_id => {} - Ok(Event::Finished(outcome @ (Outcome::Failed { .. } | Outcome::Unknown { .. }))) => { - return outcome.with_work(evidence); + let expected = definition.targets.len(); + let output = definition.output; + let (sender, receiver) = mpsc::channel(2); + let start = match remote::encode(&Input::Start(definition)) { + Ok(start) => start, + Err(message) => return failed(Stage::Admission, message.to_string()), + }; + sender.send(start).await.expect("request receiver is owned"); + let target = MachineTarget::from(&machine_id); + let mut responses = match tokio::time::timeout( + Duration::from_secs(10), + client.build_stream(&target, ReceiverStream::new(receiver)), + ) + .await + { + Ok(Ok(responses)) => responses, + Ok(Err(error)) => { + return failed( + Stage::Admission, + format!("Machine {machine_id} cannot accept the Build: {error}"), + ); + } + Err(_) => { + return failed( + Stage::Admission, + format!("Machine {machine_id} did not answer; source was not uploaded"), + ); } + }; + match tokio::time::timeout(Duration::from_secs(10), responses.message()).await { + Ok(Ok(Some(payload))) => match remote::decode::(&payload) { + Ok(Event::Admitted { + machine_id: admitted, + }) if admitted == machine_id => {} + Ok(Event::Finished(outcome @ (Outcome::Failed { .. } | Outcome::Unknown { .. }))) => { + return outcome.with_work(evidence); + } + _ => { + return failed( + Stage::Admission, + "invalid Build admission response; source was not uploaded", + ); + } + }, _ => { return failed( Stage::Admission, - "invalid Build admission response; source was not uploaded", + "Build admission was not observed; source was not uploaded", ); } - }, - _ => { - return failed( - Stage::Admission, - "Build admission was not observed; source was not uploaded", - ); } - } - progress(Progress::Stage(Stage::Upload)); - let stop = cancellation.child_token(); - let _stop_upload = stop.clone().drop_guard(); - let producer = sender.clone(); - let mut upload = tokio::task::spawn_blocking(move || { - remote::upload(inputs.root(), |frame| { - let mut payload = remote::encode(&frame)?; - loop { - if stop.is_cancelled() { - return Err("Build upload cancelled".into()); - } - match producer.try_send(payload) { - Ok(()) => return Ok(()), - Err(mpsc::error::TrySendError::Closed(_)) => { - return Err("Build upload stream closed".into()); + progress(Progress::Stage(Stage::Upload)); + let stop = cancellation.child_token(); + let _stop_upload = stop.clone().drop_guard(); + let producer = sender.clone(); + let mut upload = tokio::task::spawn_blocking(move || { + remote::upload(inputs.root(), |frame| { + let mut payload = remote::encode(&frame)?; + loop { + if stop.is_cancelled() { + return Err("Build upload cancelled".into()); + } + match producer.try_send(payload) { + Ok(()) => return Ok(()), + Err(mpsc::error::TrySendError::Closed(_)) => { + return Err("Build upload stream closed".into()); + } + Err(mpsc::error::TrySendError::Full(value)) => payload = value, } - Err(mpsc::error::TrySendError::Full(value)) => payload = value, + std::thread::sleep(Duration::from_millis(10)); } - std::thread::sleep(Duration::from_millis(10)); - } - }) - }); - let mut uploaded = false; - let mut cancelling = false; - let mut stage = Stage::Upload; - let mut deadline = - tokio::time::Instant::now() + ployz_build::EXECUTION_TIMEOUT + Duration::from_secs(70); - loop { - tokio::select! { - result = responses.message() => match result { - Ok(Some(payload)) => match remote::decode::(&payload) { - Ok(Event::Progress(event)) => { - if let Progress::Stage(observed) = &event { stage = *observed; } - evidence.observe(&event); - progress(event); + }) + }); + let mut uploaded = false; + let mut cancelling = false; + let mut stage = Stage::Upload; + let mut deadline = + tokio::time::Instant::now() + ployz_build::EXECUTION_TIMEOUT + Duration::from_secs(70); + loop { + tokio::select! { + result = responses.message() => match result { + Ok(Some(payload)) => match remote::decode::(&payload) { + Ok(Event::Progress(event)) => { + if let Progress::Stage(observed) = &event { stage = *observed; } + evidence.observe(&event); + progress(event); + } + Ok(Event::Finished(outcome)) => { + let outcome = validate_outcome(outcome, machine_id, expected, output, evidence); + if matches!(outcome, Outcome::Images { .. }) { retained = Some(responses); } + return outcome; + }, + _ => return unknown(stage, "invalid Build response; termination was not confirmed").with_work(evidence), + }, + _ => return unknown(stage, "Build stream disconnected; termination was not confirmed").with_work(evidence), + }, + result = &mut upload, if !uploaded => { + uploaded = true; + if !matches!(result, Ok(Ok(()))) && !cancelling { + cancelling = true; + deadline = tokio::time::Instant::now() + Duration::from_secs(70); + // The capture cannot be completed; ask the host to release + // admission and wait for its typed termination evidence. + if send_cancellation(&sender).await.is_err() { + return unknown(stage, "Build cancellation could not be delivered; termination was not confirmed").with_work(evidence); + } } - Ok(Event::Finished(outcome)) => return validate_outcome(outcome, machine_id, expected, output, evidence), - _ => return unknown(stage, "invalid Build response; termination was not confirmed").with_work(evidence), }, - _ => return unknown(stage, "Build stream disconnected; termination was not confirmed").with_work(evidence), - }, - result = &mut upload, if !uploaded => { - uploaded = true; - if !matches!(result, Ok(Ok(()))) && !cancelling { + () = cancellation.cancelled(), if !cancelling => { cancelling = true; deadline = tokio::time::Instant::now() + Duration::from_secs(70); - // The capture cannot be completed; ask the host to release - // admission and wait for its typed termination evidence. + // Stop the producer first so cancellation cannot sit behind an + // unbounded upload. Only two bounded frames can be in flight. if send_cancellation(&sender).await.is_err() { return unknown(stage, "Build cancellation could not be delivered; termination was not confirmed").with_work(evidence); } - } - }, - () = cancellation.cancelled(), if !cancelling => { - cancelling = true; - deadline = tokio::time::Instant::now() + Duration::from_secs(70); - // Stop the producer first so cancellation cannot sit behind an - // unbounded upload. Only two bounded frames can be in flight. - if send_cancellation(&sender).await.is_err() { - return unknown(stage, "Build cancellation could not be delivered; termination was not confirmed").with_work(evidence); - } - }, - () = tokio::time::sleep_until(deadline) => return unknown(stage, "Build response deadline expired; termination was not confirmed").with_work(evidence), + }, + () = tokio::time::sleep_until(deadline) => return unknown(stage, "Build response deadline expired; termination was not confirmed").with_work(evidence), + } } + }.await; + match outcome { + Outcome::Images { machine_id, images } => Completion::Images { + machine_id, + images, + stream: Arc::new(Mutex::new( + retained.expect("validated loaded images retain their response stream"), + )), + }, + outcome @ (Outcome::Validated { .. } + | Outcome::Published { .. } + | Outcome::Failed { .. } + | Outcome::Unknown { .. }) => Completion::Report(outcome), } } diff --git a/crates/ployz/src/compose/remote_steps.rs b/crates/ployz/src/compose/remote_steps.rs index b35e8591b5..6e4ac98d06 100644 --- a/crates/ployz/src/compose/remote_steps.rs +++ b/crates/ployz/src/compose/remote_steps.rs @@ -10,6 +10,71 @@ use ployz_core::MachineId; use tokio_util::sync::CancellationToken; impl CapturedBuild { + /// Execute this capture on one resolved Machine over the authenticated + /// Ployz stream. No local Docker or local image store is consulted. + pub async fn execute_remote( + self, + client: &crate::connect::Client, + machine_id: ployz_core::MachineId, + cancellation: tokio_util::sync::CancellationToken, + progress: impl Fn(ployz_build::Progress), + ) -> ployz_build::remote::Outcome { + if self.options.output == Output::Load { + let locations = self + .targets + .iter() + .map(|target| (target.name.clone(), machine_id)) + .collect(); + return match self + .execute_remote_steps(client, &locations, cancellation, progress) + .await + { + Ok(images) => ployz_build::remote::Outcome::Images { + machine_id, + images: images.into_iter().map(|service| service.built).collect(), + }, + Err(outcome) => outcome, + }; + } + let definition = ployz_build::remote::Definition { + retained_tags: Vec::new(), + image_contexts: Default::default(), + targets: self.targets, + output: self.options.output, + no_cache: self.options.no_cache, + pull: self.options.pull, + }; + super::super::remote_build::execute( + self.inputs, + definition, + client, + machine_id, + cancellation, + progress, + ) + .await + .into_outcome() + } + + /// Execute remotely and bind each completed output to its captured Service. + /// # Errors + /// Failed, unknown, or image-less outcomes cannot be used by Deploy. + pub async fn execute_remote_images( + self, + client: &crate::connect::Client, + machine_id: ployz_core::MachineId, + cancellation: tokio_util::sync::CancellationToken, + progress: impl Fn(ployz_build::Progress), + ) -> Result, ComposeError> { + let locations = self + .targets + .iter() + .map(|target| (target.name.clone(), machine_id)) + .collect(); + self.execute_on_machines(client, &locations, cancellation, progress) + .await + } + /// Build on already-resolved Machines, independently of application placement. /// Every target must have a location before any source is submitted. /// Keys are raw Compose Build names, not DNS-label Deploy Service names. @@ -108,6 +173,12 @@ impl CapturedBuild { let outcome = super::super::remote_build::execute( inputs, Definition { + retained_tags: self + .retained_tags + .get(&target.name) + .cloned() + .into_iter() + .collect(), targets: vec![target.clone()], image_contexts: contexts, output: Output::Load, @@ -121,7 +192,7 @@ impl CapturedBuild { ) .await; match outcome { - Outcome::Images { images, .. } => { + super::super::remote_build::Completion::Images { images, stream, .. } => { let image = images .into_iter() .next() @@ -134,9 +205,12 @@ impl CapturedBuild { machines: service.machines.clone(), location: BuildLocation::Machine(machine_id), built: image, + _retention: Some(BuildRetention::Remote { _stream: stream }), }); } - outcome @ (Outcome::Failed { .. } | Outcome::Unknown { .. }) => { + super::super::remote_build::Completion::Report( + outcome @ (Outcome::Failed { .. } | Outcome::Unknown { .. }), + ) => { if let Outcome::Failed { work: observed, .. } | Outcome::Unknown { work: observed, .. } = &outcome && let Some(evidence) = observed.0.get(&target.name) @@ -145,7 +219,9 @@ impl CapturedBuild { } return Err(outcome.with_work(work)); } - Outcome::Validated { .. } | Outcome::Published { .. } => { + super::super::remote_build::Completion::Report( + Outcome::Validated { .. } | Outcome::Published { .. } | Outcome::Images { .. }, + ) => { unreachable!("adapter validated output disposition") } } diff --git a/crates/ployz/src/deploy/pipeline.rs b/crates/ployz/src/deploy/pipeline.rs index 77035d1176..2f761d3749 100644 --- a/crates/ployz/src/deploy/pipeline.rs +++ b/crates/ployz/src/deploy/pipeline.rs @@ -293,11 +293,8 @@ pub(super) async fn push_project_images( for target in &targets { let architecture = machines.iter().find(|machine| machine.machine.id == *target) .map(|machine| machine.machine.runtime.architecture.as_str()); - let compatible = match architecture { - Some("x86_64" | "amd64") => service.built.platform == "linux/amd64", - Some("aarch64" | "arm64") => matches!(service.built.platform.as_str(), "linux/arm64" | "linux/arm64/v8"), - _ => false, - }; + let compatible = architecture.is_some_and(|architecture| + crate::image::platform_compatible(&service.built.platform, architecture)); if !compatible { return Err(Failure::usage(format!("Build for Service {} contains {}; destination Machine {target} reports architecture {}. No Service, hook, or volume change was attempted.", service.name, service.built.platform, architecture.unwrap_or("unknown")))); } diff --git a/crates/ployz/src/handlers/deploy_remote_tests.rs b/crates/ployz/src/handlers/deploy_remote_tests.rs index bf1ad46680..6cefd3aba5 100644 --- a/crates/ployz/src/handlers/deploy_remote_tests.rs +++ b/crates/ployz/src/handlers/deploy_remote_tests.rs @@ -141,19 +141,22 @@ async fn failed_unknown_and_cancelled_builds_leave_hooks_containers_and_volumes_ #[tokio::test] async fn incompatible_application_platform_refuses_before_transfer_or_mutations() { - let (root, service, builds) = fixture(); - let mutations = service.mutating_rpcs(); - let mut destination = machine('b', "application"); - destination.machine.runtime.architecture = "aarch64".into(); - let service = service.with_machines(vec![machine('a', "builder"), destination]); - let error = deploy(&root, service).await.unwrap_err().to_string(); - assert!( - error.contains("linux/amd64") && error.contains("aarch64"), - "{error}" - ); - assert_eq!(mutations.load(Ordering::SeqCst), 0); - assert!(builds.pulls.lock().unwrap().is_empty()); - fs::remove_dir_all(root).unwrap(); + for (architecture, platform) in [("aarch64", "linux/amd64"), ("armv6l", "linux/arm/v7")] { + let (root, service, builds) = fixture(); + let mutations = service.mutating_rpcs(); + *builds.platform.lock().unwrap() = Some(platform.into()); + let mut destination = machine('b', "application"); + destination.machine.runtime.architecture = architecture.into(); + let service = service.with_machines(vec![machine('a', "builder"), destination]); + let error = deploy(&root, service).await.unwrap_err().to_string(); + assert!( + error.contains(platform) && error.contains(architecture), + "{error}" + ); + assert_eq!(mutations.load(Ordering::SeqCst), 0); + assert!(builds.pulls.lock().unwrap().is_empty()); + fs::remove_dir_all(root).unwrap(); + } } #[tokio::test] @@ -352,3 +355,38 @@ async fn incomplete_build_locations_refuse_before_source_submission() { server.abort(); fs::remove_dir_all(root).unwrap(); } + +#[tokio::test] +async fn remote_deploy_accepts_matching_non_primary_architectures() { + for (architecture, platform) in [ + ("x86", "linux/386"), + ("arm", "linux/arm/v7"), + ("powerpc", "linux/ppc"), + ("powerpc64", "linux/ppc64"), + ("ppc64le", "linux/ppc64le"), + ("s390x", "linux/s390x"), + ("riscv64", "linux/riscv64"), + ("mips64el", "linux/mips64le"), + ("loongarch64", "linux/loong64"), + ] { + let (root, service, builds) = fixture(); + let created = service.created_specs(); + *builds.platform.lock().unwrap() = Some(platform.into()); + let mut destination = machine('b', "application"); + destination.machine.runtime.architecture = architecture.into(); + let yaml = fs::read_to_string(root.join("compose.yaml")).unwrap(); + fs::write( + root.join("compose.yaml"), + yaml.replace(" x-pre_deploy: {command: ['true']}\n", ""), + ) + .unwrap(); + deploy( + &root, + service.with_machines(vec![machine('a', "builder"), destination]), + ) + .await + .unwrap_or_else(|error| panic!("{architecture}/{platform}: {error}")); + assert!(!created.lock().unwrap().is_empty()); + fs::remove_dir_all(root).unwrap(); + } +} diff --git a/crates/ployz/src/image.rs b/crates/ployz/src/image.rs index 43e433100f..7a08143353 100644 --- a/crates/ployz/src/image.rs +++ b/crates/ployz/src/image.rs @@ -25,7 +25,7 @@ use self::proxy::{ImageProxy, ProxyMode, detect_mode}; mod built; mod proxy; pub use built::push_from_machine; -pub(crate) use built::{push_from_machine_using_machines, serve_build_image}; +pub(crate) use built::{platform_compatible, push_from_machine_using_machines, serve_build_image}; #[must_use] pub fn with_default_tag(image: &str) -> String { diff --git a/crates/ployz/src/image/built.rs b/crates/ployz/src/image/built.rs index 2b3ae09b7e..cf753c3948 100644 --- a/crates/ployz/src/image/built.rs +++ b/crates/ployz/src/image/built.rs @@ -121,3 +121,68 @@ pub(crate) async fn serve_build_image( .map_err(|error| ingest_error(rpc_error(error)))?; Ok(opened.destination) } + +/// Compare completed image platforms against the destination's reported architecture. +pub(crate) fn platform_compatible(platform: &str, architecture: &str) -> bool { + let (architecture, arm_generation) = match architecture { + "x86_64" => ("amd64", None), + "x86" | "i386" | "i486" | "i586" | "i686" => ("386", None), + "aarch64" => ("arm64", None), + "armv5l" => ("arm", Some(5)), + "armv6l" => ("arm", Some(6)), + "armv7l" => ("arm", Some(7)), + "armv8l" => ("arm", Some(8)), + "powerpc" => ("ppc", None), + "powerpc64" => ("ppc64", None), + "mipsel" => ("mipsle", None), + "mips64el" => ("mips64le", None), + "loongarch64" => ("loong64", None), + other => (other, None), + }; + let Some(platform) = platform.strip_prefix("linux/") else { + return false; + }; + let (image_architecture, variant) = platform + .split_once('/') + .map_or((platform, None), |(architecture, variant)| { + (architecture, Some(variant)) + }); + if image_architecture != architecture { + return false; + } + match (architecture, variant) { + (_, None) | ("amd64", Some("v1")) | ("arm64", Some("v8")) => true, + ("arm", Some(variant)) => variant + .strip_prefix('v') + .and_then(|value| value.parse::().ok()) + .is_some_and(|generation| (5..=arm_generation.unwrap_or(8)).contains(&generation)), + _ => false, + } +} + +#[cfg(test)] +mod tests { + #[test] + fn platform_compatibility_preserves_known_arm_generations() { + for (platform, architecture, compatible) in [ + ("linux/arm/v7", "armv6l", false), + ("linux/arm/v7", "armv7l", true), + ("linux/arm/v6", "armv7l", true), + ("linux/arm/v7", "arm", true), + ("linux/arm/v7/extra", "armv7l", false), + ("linux/arm64/v8", "aarch64", true), + ("linux/amd64/v3", "x86_64", false), + ("linux/386", "i686", true), + ("linux/ppc64le", "powerpc64", false), + ("linux/ppc64le", "ppc64le", true), + ("linux/mips64le", "mips64el", true), + ("linux/loong64", "loongarch64", true), + ] { + assert_eq!( + super::platform_compatible(platform, architecture), + compatible, + "{platform} on {architecture}" + ); + } + } +} diff --git a/crates/ployz/tests/build.rs b/crates/ployz/tests/build.rs index 814ae53fb0..fe2f1b4cd2 100644 --- a/crates/ployz/tests/build.rs +++ b/crates/ployz/tests/build.rs @@ -558,6 +558,7 @@ case "$1 $2" in printf '%s\n' '{{"Name":"{builder}","Nodes":[{{"Status":"running","Platforms":["linux/amd64","linux/arm64"],"DriverOpts":{{"image":"{image}"}}}}]}}' fi exit 0 ;; + 'image rm') exit 0 ;; 'image inspect') identity=$(cat "$root/store" 2>/dev/null || cat "$root/digest") media=$(cat "$root/media" 2>/dev/null || printf 'application/vnd.oci.image.manifest.v1+json') @@ -697,45 +698,3 @@ mod capture; #[path = "build/railpack.rs"] mod railpack; - -#[test] -fn deploy_binds_each_service_to_its_build_when_requested_tags_are_shared() { - use ployz::compose::BuildLocation; - let project = parse_normalized( - "services: {one: {image: 'example.test/shared:latest', build: .}, two: {image: 'example.test/shared:latest', build: .}}", - ".", - ).unwrap(); - let mut candidate = project.capture( - ployz_core::ProjectName::parse("app").unwrap(), - Default::default(), - vec![], - None, - vec![], - ); - let builds = - [("one", FIRST_CONTENT), ("two", SECOND_CONTENT)].map(|(name, digest)| BuiltService { - name: name.into(), - image: "example.test/shared:latest".into(), - machines: vec![], - location: BuildLocation::Machine(ployz_core::MachineId::parse("a".repeat(32)).unwrap()), - built: ployz_build::BuiltImage { - reference: format!("example.test/shared@{digest}"), - tags: vec!["example.test/shared:latest".into()], - platform: "linux/amd64".into(), - }, - }); - candidate.bind_builds(&builds); - for (name, digest) in [("one", FIRST_CONTENT), ("two", SECOND_CONTENT)] { - let service = candidate - .intent() - .target - .iter() - .find(|service| service.name.as_str() == name) - .unwrap(); - assert_eq!( - service.container.image, - format!("example.test/shared@{digest}") - ); - assert_eq!(service.container.pull_policy, ployz_core::PullPolicy::Never); - } -} diff --git a/crates/ployz/tests/build/capture.rs b/crates/ployz/tests/build/capture.rs index 326b2b14bb..3d044a0bfa 100644 --- a/crates/ployz/tests/build/capture.rs +++ b/crates/ployz/tests/build/capture.rs @@ -543,3 +543,47 @@ fn ssh_docker_hosts_are_refused_but_git_contexts_keep_the_captured_agent_socket( } fs::remove_dir_all(root).unwrap(); } + +#[test] +fn completed_local_build_keeps_temporary_tags_until_the_last_image_is_released() { + let root = std::env::temp_dir().join(format!("ployz-local-retention-{}", uuid::Uuid::new_v4())); + fs::create_dir_all(root.join("src")).unwrap(); + fs::write(root.join("src/Dockerfile"), "FROM scratch\n").unwrap(); + let docker = root.join("docker"); + write_docker(&docker, &root); + fs::write(root.join("digest"), FIRST_CONTENT).unwrap(); + fs::write(root.join("image"), "example.test/api:built").unwrap(); + let mut project = parse_normalized( + "services: {api: {image: 'example.test/api:built', build: ./src}}", + &root, + ) + .unwrap(); + let options = BuildOptions::default(); + let plan = plan_build(&project, &options).unwrap(); + let images = capture_build(&plan, &options, &mut project) + .unwrap() + .execute(Some(&docker)) + .unwrap(); + let clone = images.clone(); + drop(images); + assert!( + !fs::read_to_string(root.join("calls")) + .unwrap() + .lines() + .any(|call| call.starts_with("image rm")) + ); + drop(clone); + let calls = fs::read_to_string(root.join("calls")).unwrap(); + let removed = calls + .lines() + .find(|call| call.starts_with("image rm")) + .expect("temporary tag leaked"); + assert!(removed.contains(":ployz-build-"), "{removed}"); + assert!(!removed.contains("example.test/api:built"), "{removed}"); + let captured_root = fs::read_to_string(root.join("capture-root")).unwrap(); + assert!( + !Path::new(captured_root.trim()).exists(), + "capture leaked after tag cleanup" + ); + fs::remove_dir_all(root).unwrap(); +} diff --git a/crates/ployz/tests/build_layer3/remote.rs b/crates/ployz/tests/build_layer3/remote.rs index f1716d5498..1e7f995410 100644 --- a/crates/ployz/tests/build_layer3/remote.rs +++ b/crates/ployz/tests/build_layer3/remote.rs @@ -215,6 +215,7 @@ async fn request( sender .send( remote::encode(&Input::Start(Definition { + retained_tags: Vec::new(), image_contexts: Default::default(), targets: vec![ployz_build::Target { name: "app".into(), @@ -315,6 +316,7 @@ async fn remote_build_delivers_dependency_content_and_deploys_without_a_registry String::from_utf8_lossy(&output.stderr) ); assert!(!root.join("local-docker-called").exists()); + assert_temporary_tags_released(&cluster).await; let container = cluster .machine_shell(0, "docker ps -q --filter label=ployz.service.name=app") .unwrap(); @@ -394,5 +396,27 @@ async fn remote_build_delivers_dependency_content_and_deploys_without_a_registry .unwrap(), "dependency-output\nremote-application-ran\n" ); + drop(images); + assert_temporary_tags_released(&cluster).await; fs::remove_dir_all(root).unwrap(); } + +async fn assert_temporary_tags_released(cluster: &Cluster) { + tokio::time::timeout(Duration::from_secs(10), async { + loop { + let mut empty = true; + for machine in 0..2 { + let tags = cluster + .machine_shell(machine, "docker image ls --format '{{.Tag}}'") + .unwrap(); + empty &= !tags.lines().any(|tag| tag.starts_with("ployz-build-")); + } + if empty { + return; + } + tokio::time::sleep(Duration::from_millis(100)).await; + } + }) + .await + .expect("command left temporary Build tags on a Machine"); +} diff --git a/crates/ployz/tests/support/build.rs b/crates/ployz/tests/support/build.rs index f08a01f4dd..b56ff81caa 100644 --- a/crates/ployz/tests/support/build.rs +++ b/crates/ployz/tests/support/build.rs @@ -17,6 +17,7 @@ use tonic::{Request, Response, Status}; #[derive(Default)] pub struct BuildFixture { pub terminal: Mutex>, + pub platform: Mutex>, pub definitions: Mutex>, pub stores: Mutex>, pub opened: Mutex>, @@ -118,7 +119,12 @@ impl BuildFixture { (offset + index + 1).to_string().repeat(64) ), tags: vec!["registry.invalid/shared:latest".into()], - platform: "linux/amd64".into(), + platform: self + .platform + .lock() + .unwrap() + .clone() + .unwrap_or_else(|| "linux/amd64".into()), }) .collect::>(); let mut stores = self.stores.lock().unwrap(); diff --git a/crates/ployzd/Cargo.toml b/crates/ployzd/Cargo.toml index f0546ca1a9..6a2ed0b238 100644 --- a/crates/ployzd/Cargo.toml +++ b/crates/ployzd/Cargo.toml @@ -24,7 +24,7 @@ http-body = "1.1.0" http-body-util = "0.1.5" instant-acme = "0.8.5" listenfd = "1.0.2" -nix = { version = "0.29.0", default-features = false, features = ["fs"] } +nix = { version = "0.29.0", default-features = false, features = ["fs", "feature"] } ipnet = "2.12.1" ployz-core.workspace = true ployz-build.workspace = true diff --git a/crates/ployzd/src/build.rs b/crates/ployzd/src/build.rs index 3f1b0afbe0..eff90dfa5b 100644 --- a/crates/ployzd/src/build.rs +++ b/crates/ployzd/src/build.rs @@ -6,11 +6,21 @@ use ployz_build::{ remote::{self, Definition, Event, Input, Outcome, Upload}, }; use ployz_core::{MachineId, OpaquePayload}; -use std::time::Duration; +use std::{ + sync::{Arc, Mutex}, + time::Duration, +}; + use tokio::sync::mpsc; use tokio_stream::{Stream, StreamExt, wrappers::ReceiverStream}; use tonic::Status; +#[derive(Default)] +struct AttemptState { + evidence: Mutex, + retained: Mutex>, +} + pub(crate) fn start( machine_id: MachineId, requests: impl Stream> + Send + Unpin + 'static, @@ -18,7 +28,8 @@ pub(crate) fn start( ) -> RpcStream { let (events, receiver) = mpsc::channel(8); tokio::spawn(async move { - let outcome = attempt(machine_id, requests, &events, policy).await; + let state = Arc::new(AttemptState::default()); + let outcome = attempt(machine_id, requests, &events, policy, state.clone()).await; let message = "Build terminal report exceeds the response size limit"; let fallback = match &outcome { Outcome::Unknown { stage, .. } => Outcome::Unknown { @@ -35,6 +46,17 @@ pub(crate) fn start( remote::encode(&Event::Finished(fallback)).expect("bounded terminal fallback") }); let _ = tokio::time::timeout(Duration::from_secs(5), events.send(Ok(payload))).await; + // Admission is already released. Only image retention follows this stream. + if state + .retained + .lock() + .expect("Build retention lock") + .is_some() + { + events.closed().await; + } + let retained = state.retained.lock().expect("Build retention lock").take(); + let _ = tokio::task::spawn_blocking(move || drop(retained)).await; }); ReceiverStream::new(receiver) } @@ -44,6 +66,7 @@ async fn attempt( mut requests: impl Stream> + Send + Unpin + 'static, events: &mpsc::Sender>, policy: HostPolicy, + state: Arc, ) -> Outcome { let start = tokio::time::timeout(Duration::from_secs(10), requests.next()).await; let definition = match start { @@ -72,10 +95,9 @@ async fn attempt( Ok(Err(error)) => return failure(Stage::Admission, error), Err(_) => return failed(Stage::Admission, "Build admission task failed"), }; - let evidence = std::sync::Arc::new(std::sync::Mutex::new(ployz_build::WorkEvidence::new( - &definition.targets, - ))); - let observed = evidence.clone(); + *state.evidence.lock().expect("Build evidence lock") = + ployz_build::WorkEvidence::new(&definition.targets); + let observed = state.clone(); let cancellation = admission.cancellation(); let remaining = admission.remaining(); let deadline = tokio::time::Instant::now() + remaining; @@ -97,7 +119,7 @@ async fn attempt( let input = pump(&mut requests, upload); tokio::pin!(input); tokio::select! { - result = &mut execution => return joined(result).with_work(evidence.lock().expect("Build evidence lock").clone()), + result = &mut execution => return joined(result).with_work(state.evidence.lock().expect("Build evidence lock").clone()), reason = &mut input => reason, () = events.closed() => "Build client disconnected".to_owned(), () = tokio::time::sleep_until(deadline) => "Build active timeout expired".to_owned(), @@ -126,7 +148,7 @@ async fn attempt( ), }, }; - outcome.with_work(evidence.lock().expect("Build evidence lock").clone()) + outcome.with_work(state.evidence.lock().expect("Build evidence lock").clone()) } async fn pump( @@ -159,7 +181,7 @@ fn receive_and_execute( admission: Admission, policy: HostPolicy, events: &mpsc::Sender>, - evidence: &std::sync::Mutex, + state: &AttemptState, ) -> Outcome { let cancellation = admission.cancellation(); let mut upload = match Upload::new() { @@ -194,7 +216,8 @@ fn receive_and_execute( }; let deadline = std::time::Instant::now() + admission.remaining(); let progress = |progress| { - evidence + state + .evidence .lock() .expect("Build evidence lock") .observe(&progress); @@ -232,12 +255,21 @@ fn receive_and_execute( context.source.port = proxy.port; proxies.push(proxy); } - match upload.execute(&definition, admission, Some(&policy.docker), &progress) { - Ok(images) => match definition.output { - Output::Load => Outcome::Images { machine_id, images }, - Output::Validate => Outcome::Validated { machine_id }, - Output::Registry => Outcome::Published { machine_id }, - }, + let result = upload.execute(&definition, admission, Some(&policy.docker), &progress); + match result { + Ok(completed) => { + if !definition.retained_tags.is_empty() { + *state.retained.lock().expect("Build retention lock") = Some(completed.retention); + } + match definition.output { + Output::Load => Outcome::Images { + machine_id, + images: completed.images, + }, + Output::Validate => Outcome::Validated { machine_id }, + Output::Registry => Outcome::Published { machine_id }, + } + } Err(error) => failure(error.stage(), error), } } diff --git a/crates/ployzd/src/build/tests.rs b/crates/ployzd/src/build/tests.rs index 1109a8bb6a..0f439c1466 100644 --- a/crates/ployzd/src/build/tests.rs +++ b/crates/ployzd/src/build/tests.rs @@ -94,6 +94,7 @@ impl Fixture { sender .send( remote::encode(&Input::Start(Definition { + retained_tags: Vec::new(), image_contexts: Default::default(), targets: vec![ployz_build::Target { name: "api".into(), @@ -316,6 +317,7 @@ case "$1 $2" in previous=$arg done if [ -f "$root/fail-after-output" ]; then exit 1; fi ;; + 'image rm') printf '%s\n' "$*" >> "$root/released-tags" ;; 'image inspect') printf '%s\n' '{{"Os":"linux","Architecture":"amd64","Descriptor":{{"mediaType":"application/vnd.oci.image.manifest.v1+json","digest":"sha256:{}"}}}}' ;; *) exit 1 ;; esac @@ -376,6 +378,7 @@ async fn upload_timeout_stops_before_execution_and_releases_admission() { let (sender, receiver) = mpsc::channel(2); sender .send(Ok(remote::encode(&Input::Start(Definition { + retained_tags: Vec::new(), image_contexts: Default::default(), targets: vec![ployz_build::Target { name: "api".into(), @@ -555,3 +558,37 @@ async fn host_configuration_and_cache_clearing_share_remote_admission() { )); assert!(Admission::try_acquire_with(&fixture.policy).is_ok()); } + +#[tokio::test] +async fn dropping_completed_build_releases_only_its_temporary_tags() { + let fixture = Fixture::new().await; + let client = ployz::connect::connect( + Path::new("/missing-test-config"), + Some(&fixture.address.replace("http://", "tcp://")), + None, + ) + .await + .unwrap(); + let images = fixture + .capture() + .execute_remote_images( + &client, + fixture.machine.id, + tokio_util::sync::CancellationToken::new(), + |_| {}, + ) + .await + .unwrap(); + assert!(!fixture.root.join("released-tags").exists()); + drop(images); + tokio::time::timeout(Duration::from_secs(3), async { + while !fixture.root.join("released-tags").exists() { + tokio::time::sleep(Duration::from_millis(20)).await; + } + }) + .await + .expect("completed Build leaked its retention tag"); + let released = fs::read_to_string(fixture.root.join("released-tags")).unwrap(); + assert!(released.contains(":ployz-build-"), "{released}"); + assert!(!released.contains("example.test/api:built"), "{released}"); +} diff --git a/crates/ployzd/src/machine/mod.rs b/crates/ployzd/src/machine/mod.rs index 645d8b010b..5e5e018fa9 100644 --- a/crates/ployzd/src/machine/mod.rs +++ b/crates/ployzd/src/machine/mod.rs @@ -667,7 +667,11 @@ pub fn local_runtime() -> MachineRuntime { daemon_version: env!("CARGO_PKG_VERSION").into(), docker_version: docker_version(Path::new("docker"), DOCKER_VERSION_TIMEOUT), hostname: read_trimmed("/etc/hostname"), - architecture: std::env::consts::ARCH.into(), + // Kernel architecture preserves distinctions lost by Rust's ARCH, + // including ppc64/ppc64le and ARM generations. + architecture: nix::sys::utsname::uname() + .map(|system| system.machine().to_string_lossy().into_owned()) + .unwrap_or_else(|_| std::env::consts::ARCH.into()), os_pretty_name: fs::read_to_string("/etc/os-release") .ok() .and_then(|contents| { From 6e32e1c12d51d561b7d7de7d3d20073209a614c9 Mon Sep 17 00:00:00 2001 From: Codex Date: Wed, 9 Sep 2026 04:22:13 +0000 Subject: [PATCH 3/3] Let BuildKit validate cross-platform named image contexts --- crates/ployz-build/src/lib.rs | 22 --------- crates/ployz-build/src/tests.rs | 83 +++++++++++++++++++++++++++++++++ 2 files changed, 83 insertions(+), 22 deletions(-) diff --git a/crates/ployz-build/src/lib.rs b/crates/ployz-build/src/lib.rs index 54738eb1af..f1d30c58c6 100644 --- a/crates/ployz-build/src/lib.rs +++ b/crates/ployz-build/src/lib.rs @@ -334,28 +334,6 @@ pub fn execute_admitted( .map_err(|error| error.at(Stage::Preparation))?; preparation = railpack::prepare(&docker, request, &native, &admission.resources) .map_err(|error| error.at(Stage::Preparation))?; - for image in request.image_contexts.values() { - for target in request.targets { - let platforms = if target.platforms.is_empty() { - vec![native.clone()] - } else { - target.platforms.clone() - }; - for platform in platforms { - if !image - .platforms - .iter() - .any(|observed| covers(observed, &platform)) - { - return Err(BuildError::Request(format!( - "image context {} contains {:?}, but this Build requires {platform}", - image.reference, image.platforms - )) - .at(Stage::Preparation)); - } - } - } - } let overrides = preparation .as_ref() .map(railpack::Preparation::override_file); diff --git a/crates/ployz-build/src/tests.rs b/crates/ployz-build/src/tests.rs index 5d8f50169e..2357dfbf02 100644 --- a/crates/ployz-build/src/tests.rs +++ b/crates/ployz-build/src/tests.rs @@ -469,3 +469,86 @@ exit 0 } std::fs::remove_dir_all(directory).unwrap(); } + +#[test] +fn cross_platform_image_context_reaches_buildkit() { + use std::os::unix::fs::DirBuilderExt as _; + + let directory = std::env::temp_dir().join(format!("ployz-context-{}", uuid::Uuid::new_v4())); + std::fs::DirBuilder::new() + .mode(0o700) + .create(&directory) + .unwrap(); + let program = directory.join("docker"); + executable( + &program, + &format!( + r#"#!/bin/sh +case "$1 $2" in + 'context show') echo default ;; + 'info --format') echo '{{"DriverStatus":[["driver-type","io.containerd.snapshotter.v1"]],"Architecture":"arm64","OSType":"linux"}}' ;; + 'buildx ls') echo '{{"Name":"{}","Nodes":[{{"Status":"running","Platforms":["linux/arm64"]}}]}}' ;; + 'buildx bake') printf '%s\n' "$@" > bake-arguments ;; +esac +exit 0 +"#, + builder_name() + ), + ); + std::fs::write( + directory.join("compose.yaml"), + "services: {app: {build: {context: ., additional_contexts: {base: 'service:base'}}}}\n", + ) + .unwrap(); + std::fs::write( + directory.join("Dockerfile"), + "FROM --platform=linux/amd64 base\n", + ) + .unwrap(); + let contexts = BTreeMap::from([( + "base".into(), + ImageContext { + reference: format!("example.test/base@sha256:{}", "1".repeat(64)), + platforms: vec!["linux/amd64".into()], + source: ployz_core::ImageIngestDestination { + management_address: ployz_core::ManagementAddress( + std::net::Ipv4Addr::LOCALHOST.to_ipv6_mapped(), + ), + port: 5000, + }, + }, + )]); + let environment = BTreeMap::new(); + let targets = [target("app", Some("linux/arm64"))]; + let request = Request { + image_contexts: &contexts, + compose_file: Path::new("compose.yaml"), + working_dir: &directory, + environment: &environment, + docker: Some(&program), + targets: &targets, + railpack: &[], + build_args: &[], + output: Output::Validate, + no_cache: false, + pull: false, + }; + let admission = Admission::try_acquire_with(&HostPolicy { + state_directory: directory.clone(), + configuration_file: directory.join("build.yaml"), + ..Default::default() + }) + .unwrap(); + execute_admitted(&request, admission, &|_| {}).unwrap(); + assert!( + std::fs::read_to_string(directory.join("bake-arguments")) + .unwrap() + .contains("app.platform=linux/arm64") + ); + assert!( + std::fs::read_to_string(directory.join("compose.yaml")) + .unwrap() + .contains("docker-image://127.0.0.1:5000/example.test/base@sha256:") + ); + std::fs::remove_dir_all(directory).unwrap(); +}