diff --git a/crates/ployz-build/src/execution.rs b/crates/ployz-build/src/execution.rs index 1d645d0b09..f2856f6427 100644 --- a/crates/ployz-build/src/execution.rs +++ b/crates/ployz-build/src/execution.rs @@ -228,12 +228,14 @@ impl Admission { #[cfg(test)] mod tests { use super::*; + use std::os::unix::fs::PermissionsExt; #[test] fn restart_removes_abandoned_uploads_but_keeps_unknown_ownership_unavailable() { let root = std::env::temp_dir().join(format!("ployz-build-restart-{}", uuid::Uuid::new_v4())); std::fs::create_dir(&root).unwrap(); + std::fs::set_permissions(&root, std::fs::Permissions::from_mode(0o700)).unwrap(); let policy = HostPolicy { state_directory: root.clone(), docker: root.join("docker"), @@ -302,6 +304,7 @@ mod tests { fn expired_admission_and_docker_report_configured_budget() { let root = std::env::temp_dir().join(format!("ployz-budget-{}", uuid::Uuid::new_v4())); std::fs::create_dir(&root).unwrap(); + std::fs::set_permissions(&root, std::fs::Permissions::from_mode(0o700)).unwrap(); let policy = HostPolicy { state_directory: root.clone(), active_timeout: Duration::from_secs(5), @@ -332,6 +335,7 @@ mod tests { let root = std::env::temp_dir().join(format!("ployz-upload-cancel-{}", uuid::Uuid::new_v4())); std::fs::create_dir_all(root.join("build-upload")).unwrap(); + std::fs::set_permissions(&root, std::fs::Permissions::from_mode(0o700)).unwrap(); std::fs::write(root.join("build-upload/existing"), "capture").unwrap(); let policy = HostPolicy { state_directory: root.clone(), 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 bbf628a606..f1d30c58c6 100644 --- a/crates/ployz-build/src/lib.rs +++ b/crates/ployz-build/src/lib.rs @@ -12,10 +12,13 @@ mod builder; mod execution; +mod image_contexts; mod index; mod policy; pub use policy::clear_cache; mod received_recipe; +mod retention; +pub use retention::{ImageRetention, RetainedImages}; pub mod remote; mod upload; @@ -61,6 +64,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. @@ -121,6 +126,39 @@ pub struct BuiltImage { pub location: String, } +impl BuiltImage { + /// Name this manifest in the repository used by its Build. + /// # Errors + /// Rejects missing or malformed repository evidence. + pub fn repository_reference(&self) -> Result { + let tag = self + .tags + .first() + .ok_or_else(|| BuildError::Result("Build image has no tags".into()))?; + let reference = tag + .parse::() + .map_err(|error| BuildError::Result(error.to_string()))?; + Ok(format!( + "{}/{}@{}", + reference.registry(), + reference.repository(), + self.reference + )) + } +} + +/// 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, + /// Platforms actually available in that completed image. + pub platforms: Vec, + /// 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 { @@ -277,6 +315,7 @@ pub fn execute_admitted( docker .require_local() .and_then(|()| docker.run("check Buildx", &["buildx", "version"], Streams::Captured)) + .and_then(|_| image_contexts::prepare(request)) .map_err(|error| { let error = error.at(Stage::Preparation); if !error.is_unknown() diff --git a/crates/ployz-build/src/received_recipe.rs b/crates/ployz-build/src/received_recipe.rs index d9187a885e..4f39d35a92 100644 --- a/crates/ployz-build/src/received_recipe.rs +++ b/crates/ployz-build/src/received_recipe.rs @@ -39,6 +39,39 @@ pub fn validate_capture( { 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 + || (context.platforms.is_empty() + || context + .platforms + .iter() + .any(|platform| !crate::remote::linux_platform(platform))) + { + 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() + .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")?; @@ -49,6 +82,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) { @@ -101,7 +145,7 @@ pub fn validate_capture( .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) { @@ -119,7 +163,7 @@ pub fn validate_capture( 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) => { @@ -127,7 +171,7 @@ pub fn validate_capture( 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 => {} @@ -448,6 +492,8 @@ mod tests { ) .unwrap(); let definition = Definition { + retained_tags: Vec::new(), + image_contexts: Default::default(), targets: vec![crate::Target { name: "api".into(), platforms: Vec::new(), @@ -493,6 +539,8 @@ 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(), platforms: Vec::new(), diff --git a/crates/ployz-build/src/remote.rs b/crates/ployz-build/src/remote.rs index 7a6f5580f2..42dcbe3eda 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::{AdmittedUpload, 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,12 @@ 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, /// Targets in this admitted attempt. pub targets: Vec, /// Requested disposition of completed output. @@ -138,6 +160,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 requests_cannot_supply_execution_host_policy() { for setting in [ 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/tests.rs b/crates/ployz-build/src/tests.rs index 100db50a0b..2357dfbf02 100644 --- a/crates/ployz-build/src/tests.rs +++ b/crates/ployz-build/src/tests.rs @@ -195,6 +195,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, @@ -275,7 +276,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"), @@ -368,6 +371,7 @@ esac let environment = BTreeMap::new(); let targets = [target("api", None)]; let request = Request { + image_contexts: &BTreeMap::new(), compose_file: Path::new("compose.yaml"), working_dir: &directory, environment: &environment, @@ -424,6 +428,7 @@ fn docker_preflight_holds_quarantine_and_clears_confirmed_failures() { let targets = [target("api", None)]; let environment = BTreeMap::new(); let request = Request { + image_contexts: &BTreeMap::new(), railpack: &[], compose_file: &directory.join("compose.json"), working_dir: &directory, @@ -464,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(); +} diff --git a/crates/ployz-build/src/upload.rs b/crates/ployz-build/src/upload.rs index 36b5ced5c7..596aa5cc68 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::{ @@ -80,7 +80,7 @@ impl AdmittedUpload { definition: &Definition, docker: Option<&Path>, progress: &(dyn Fn(Progress) + Sync), - ) -> Result, BuildError> { + ) -> Result { let _ownership = self.admission.lock.clone(); self.upload .complete() @@ -271,7 +271,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. @@ -281,7 +281,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()))?; @@ -290,8 +290,14 @@ impl CompletedUpload { .map_err(|error| BuildError::Request(io_error(error).to_string()))?; } let environment = environment(&upload.root); + let retention = crate::ImageRetention::new( + definition.retained_tags.clone(), + docker, + environment.clone(), + )?; let result = crate::execute_admitted( &Request { + image_contexts: &definition.image_contexts, railpack: &railpack, compose_file: Path::new("compose.yaml"), working_dir: &upload.root, @@ -309,7 +315,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-build/tests/host_policy.rs b/crates/ployz-build/tests/host_policy.rs index 20c7179330..60428eb924 100644 --- a/crates/ployz-build/tests/host_policy.rs +++ b/crates/ployz-build/tests/host_policy.rs @@ -159,6 +159,7 @@ exit 0 refresh_cache: false, }]; let request = Request { + image_contexts: &BTreeMap::new(), compose_file: &host.file("compose.yaml"), working_dir: &host.policy.state_directory, environment: &BTreeMap::from([("PATH".into(), "/usr/bin:/bin".into())]), @@ -186,6 +187,7 @@ fn resource_launch_failure_reports_preparation_and_releases_confirmed_ownership( }]; let error = execute_admitted( &Request { + image_contexts: &BTreeMap::new(), compose_file: &host.file("compose.yaml"), working_dir: &host.policy.state_directory, environment: &BTreeMap::new(), @@ -316,6 +318,7 @@ echo "$*" >> "$root/mutations" ]); let error = execute_admitted( &Request { + image_contexts: &BTreeMap::new(), compose_file: &host.file("compose.yaml"), working_dir: &host.policy.state_directory, environment: &environment, diff --git a/crates/ployz/src/cli/mod.rs b/crates/ployz/src/cli/mod.rs index 5b0bc49c0d..84ca9ea081 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 a3b2ebc0e0..5faf8586ac 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}; @@ -10,6 +11,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 +37,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, @@ -69,17 +71,59 @@ pub struct CapturedBuild { options: BuildOptions, environment: BTreeMap, inputs: BuildInputs, + retained_tags: BTreeMap, +} + +/// 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)] +#[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. + 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. 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 { @@ -142,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(); @@ -199,6 +244,27 @@ pub fn capture_build( } targets.push(ployz_build::Target { name, platforms }); 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() + ); + 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())])); + 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) @@ -329,6 +395,7 @@ pub fn capture_build( .map(|(key, value)| (key.clone(), value.clone())) .collect(), inputs, + retained_tags, }) } @@ -544,32 +611,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 { - let definition = ployz_build::remote::Definition { - 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 - } - /// Build this capture through the shared runner, without reading the /// original sources again. /// @@ -598,8 +639,19 @@ 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"), working_dir: self.inputs.root(), environment: &environment, @@ -629,9 +681,12 @@ 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, + _retention: Some(retention.clone()), }) .collect()) } @@ -853,7 +908,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)?; @@ -871,6 +925,7 @@ fn include_service<'a>( )?; } } + selected.push(name); visiting.remove(name); seen.insert(name); Ok(()) @@ -949,6 +1004,8 @@ mod tests { let plan = plan_build(&project, &options).unwrap(); let captured = capture_build(&plan, &options, &mut project).unwrap(); let definition = ployz_build::remote::Definition { + image_contexts: Default::default(), + retained_tags: Vec::new(), targets: captured.targets, output: options.output, no_cache: options.no_cache, diff --git a/crates/ployz/src/compose/build_inputs.rs b/crates/ployz/src/compose/build_inputs.rs index c7569340c7..daa1f3a689 100644 --- a/crates/ployz/src/compose/build_inputs.rs +++ b/crates/ployz/src/compose/build_inputs.rs @@ -124,6 +124,51 @@ 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 metadata = inputs.root.join("private/railpack.json"); + if metadata.exists() { + let mut recipes: Vec = + serde_json::from_slice(&fs::read(&metadata).map_err(input_error)?) + .map_err(|error| ComposeError::Invalid(error.to_string()))?; + recipes.retain(|recipe| recipe.name == name); + fs::remove_file(metadata).map_err(input_error)?; + inputs.railpack(&recipes)?; + } + 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 @@ -629,6 +674,43 @@ fn validate_link(path: &Path, root: &Path, selection: Option<&Selection>) -> io: #[cfg(test)] mod tests { use super::*; + + #[test] + fn per_service_capture_keeps_only_its_railpack_recipe() { + let inputs = BuildInputs::new().unwrap(); + inputs + .compose("services: {api: {build: .}, web: {build: .}, base: {build: .}}\n") + .unwrap(); + inputs + .railpack(&["api", "web"].map(|name| ployz_build::Railpack { + name: name.into(), + context: "source".into(), + variables: BTreeMap::new(), + refresh_cache: false, + })) + .unwrap(); + for name in ["api", "web", "base"] { + let selected = inputs.for_service(name).unwrap(); + let metadata = selected.root().join("private/railpack.json"); + if name == "base" { + assert!(!metadata.exists()); + } else { + let recipes: Vec = + serde_json::from_slice(&fs::read(metadata).unwrap()).unwrap(); + assert_eq!(recipes.len(), 1); + assert_eq!(recipes.first().unwrap().name, name); + } + } + let original: Vec = + serde_json::from_slice(&fs::read(inputs.root().join("private/railpack.json")).unwrap()) + .unwrap(); + assert_eq!( + original.len(), + 2, + "per-target selection changed the original capture" + ); + } + use std::os::unix::net::UnixListener; #[test] diff --git a/crates/ployz/src/compose/capture.rs b/crates/ployz/src/compose/capture.rs index d473c662b1..4671f04416 100644 --- a/crates/ployz/src/compose/capture.rs +++ b/crates/ployz/src/compose/capture.rs @@ -30,6 +30,29 @@ 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], + ) -> Result<(), super::ComposeError> { + for service in &mut self.intent.target { + if let Some(build) = builds + .iter() + .find(|build| build.name == service.name.as_str()) + { + service.container.image = build.built.repository_reference().map_err(|error| { + super::ComposeError::Build { + services: build.name.clone(), + source: error, + } + })?; + service.container.pull_policy = ployz_core::PullPolicy::Never; + } + } + Ok(()) + } + /// Identity of this capture, unchanged by later edits to the source files. #[must_use] pub fn id(&self) -> &str { @@ -86,3 +109,57 @@ 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: digest.into(), + tags: vec!["example.test/shared:latest".into()], + platforms: vec!["linux/amd64".into()], + location: "unix:///var/run/docker.sock".into(), + }, + }); + candidate.bind_builds(&builds).unwrap(); + 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/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 9d8a8e3071..2c3ea54dd9 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,7 +43,7 @@ pub(super) async fn execute( machine_id: MachineId, cancellation: CancellationToken, progress: impl Fn(Progress), -) -> Outcome { +) -> Completion { let mut queued = None; let mut executing = None; let outcome = execute_attempt( @@ -63,132 +86,152 @@ async fn execute_attempt( machine_id: MachineId, cancellation: CancellationToken, mut progress: impl FnMut(Progress), -) -> Outcome { - let mut evidence = ployz_build::WorkEvidence::new(&definition.targets); - 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 let Err(error) = remote::validate_capture(inputs.root(), &definition) { + return failed(Stage::Preparation, error.to_string()).with_work(evidence); } - Err(_) => { - return failed( - Stage::Admission, - format!("Machine {machine_id} did not answer; source was not uploaded"), - ); - } - }; - let waiting_since = tokio::time::Instant::now(); - let mut admission_deadline = waiting_since + Duration::from_secs(10); - let mut queued = false; - let active_timeout = loop { - tokio::select! { - biased; - () = cancellation.cancelled() => { - // No source has left this client. Dropping the stream removes - // its connection-scoped waiter, even if admission raced cancellation. - return failed(Stage::Queued, "Build cancelled before upload; execution was not attempted").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}"), + ); } - result = responses.message() => match result { - Ok(Some(payload)) => match remote::decode::(&payload) { - Ok(Event::Admitted { machine_id: admitted, active_timeout }) - if admitted == machine_id && !active_timeout.is_zero() && active_timeout <= Duration::from_secs(86400) => break active_timeout, - Ok(Event::Progress(Progress::Stage(Stage::Queued))) if !queued => { - queued = true; - // The Machine's validated local policy is bounded at 24h. - admission_deadline = waiting_since + Duration::from_secs(86410); - progress(Progress::Stage(Stage::Queued)); - } - 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").with_work(evidence), - }, - _ => return failed(Stage::Admission, "Build admission was not observed; source was not uploaded").with_work(evidence), - }, - () = tokio::time::sleep_until(admission_deadline) => return failed(Stage::Queued, "Build admission deadline expired; source was not uploaded").with_work(evidence), - } - }; - 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()); + Err(_) => { + return failed( + Stage::Admission, + format!("Machine {machine_id} did not answer; source was not uploaded"), + ); + } + }; + let waiting_since = tokio::time::Instant::now(); + let mut admission_deadline = waiting_since + Duration::from_secs(10); + let mut queued = false; + let active_timeout = loop { + tokio::select! { + biased; + () = cancellation.cancelled() => { + // No source has left this client. Dropping the stream removes + // its connection-scoped waiter, even if admission raced cancellation. + return failed(Stage::Queued, "Build cancelled before upload; execution was not attempted").with_work(evidence); } - match producer.try_send(payload) { - Ok(()) => return Ok(()), - Err(mpsc::error::TrySendError::Closed(_)) => { - return Err("Build upload stream closed".into()); + result = responses.message() => match result { + Ok(Some(payload)) => match remote::decode::(&payload) { + Ok(Event::Admitted { machine_id: admitted, active_timeout }) + if admitted == machine_id && !active_timeout.is_zero() && active_timeout <= Duration::from_secs(86400) => break active_timeout, + Ok(Event::Progress(Progress::Stage(Stage::Queued))) if !queued => { + queued = true; + // The Machine's validated local policy is bounded at 24h. + admission_deadline = waiting_since + Duration::from_secs(86410); + progress(Progress::Stage(Stage::Queued)); + } + 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").with_work(evidence), + }, + _ => return failed(Stage::Admission, "Build admission was not observed; source was not uploaded").with_work(evidence), + }, + () = tokio::time::sleep_until(admission_deadline) => return failed(Stage::Queued, "Build admission deadline expired; source was not uploaded").with_work(evidence), + } + }; + 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() + active_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() + active_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), } } @@ -227,7 +270,7 @@ fn validate_outcome( || image .platforms .iter() - .any(|platform| !linux_platform(platform)) + .any(|platform| !remote::linux_platform(platform)) }) { return unknown( Stage::Output, @@ -257,19 +300,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..67eca93644 --- /dev/null +++ b/crates/ployz/src/compose/remote_steps.rs @@ -0,0 +1,259 @@ +//! 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 { + /// 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. + /// # 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 + .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, + &cancellation, + ) + .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.repository_reference().map_err(|error| { + failed(Stage::Preparation, error.to_string()).with_work(work.clone()) + })?, + platforms: image.built.platforms.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 { + retained_tags: self + .retained_tags + .get(&target.name) + .cloned() + .into_iter() + .collect(), + 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 { + super::super::remote_build::Completion::Images { images, stream, .. } => { + 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, + _retention: Some(BuildRetention::Remote { _stream: stream }), + }); + } + 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) + { + work.0.insert(target.name.clone(), evidence.clone()); + } + return Err(outcome.with_work(work)); + } + super::super::remote_build::Completion::Report( + Outcome::Validated { .. } | Outcome::Published { .. } | Outcome::Images { .. }, + ) => { + 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 483e17f6c0..d6e991bca0 100644 --- a/crates/ployz/src/deploy/apply.rs +++ b/crates/ployz/src/deploy/apply.rs @@ -163,7 +163,12 @@ pub(crate) async fn deploy_project( ) -> Result<(), Failure> { let machines = crate::cancellation::read(cancellation, async { Ok(client.machines().await?) }).await?; - let outcome = push_project_images(client, builds, &machines, cancellation).await?; + let preview = crate::cancellation::read( + cancellation, + plan_project(client, candidate, machines.clone()), + ) + .await?; + let outcome = push_project_images(client, builds, &machines, &preview, cancellation).await?; print_pushed_images(&outcome); if !outcome.failures.is_empty() { return Err(Failure::usage(format!( @@ -171,8 +176,6 @@ pub(crate) async fn deploy_project( outcome.failures.join("; ") ))); } - let preview = - crate::cancellation::read(cancellation, plan_project(client, candidate, machines)).await?; println!("Captured candidate {}", candidate.id()); print_warnings(&preview); confirm_and_execute(client, &preview, gate, cancellation).await diff --git a/crates/ployz/src/deploy/pipeline.rs b/crates/ployz/src/deploy/pipeline.rs index fbc6a86629..ecb74b06cf 100644 --- a/crates/ployz/src/deploy/pipeline.rs +++ b/crates/ployz/src/deploy/pipeline.rs @@ -281,12 +281,32 @@ pub(super) async fn push_project_images( client: &mut Client, builds: &[BuiltService], machines: &[MachineObservation], + preview: &DeployPlan, cancellation: &CancellationToken, ) -> Result { let mut pushed = Vec::new(); let mut failures = Vec::new(); - for service in builds { - match push_image(client, service, machines, cancellation).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 = architecture.is_some_and(|architecture| + service.built.platforms.iter().any(|platform| crate::image::platform_compatible(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.platforms.join(", "), 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, cancellation).await { Ok((images, service_failures)) => { pushed.extend(images); failures.extend(service_failures); @@ -529,23 +549,42 @@ async fn push_image( client: &mut Client, service: &BuiltService, machines: &[MachineObservation], + targets: &[String], cancellation: &CancellationToken, ) -> 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, - cancellation, - ) - .await?; + let result = match service.location { + crate::compose::BuildLocation::Local => { + // Retain each digest even when multiple Services requested one tag. + let tag = service + .built + .repository_reference() + .map_err(|error| crate::image::PushError::InvalidReference { + reference: service.built.reference.clone(), + message: error.to_string(), + })? + .replace("@sha256:", ":ployz-sha256-"); + crate::image::push_using_machines( + client, + crate::image::ImageContent::built(&tag, &service.built.reference), + None, + targets, + machines, + cancellation, + ) + .await? + } + crate::compose::BuildLocation::Machine(source) => { + crate::image::push_from_machine_using_machines( + client, + &service.built, + source, + targets, + machines, + cancellation, + ) + .await? + } + }; let pushed = result .successes .iter() diff --git a/crates/ployz/src/handlers/build.rs b/crates/ployz/src/handlers/build.rs index 11cf6085df..79ca86eb6c 100644 --- a/crates/ployz/src/handlers/build.rs +++ b/crates/ployz/src/handlers/build.rs @@ -71,43 +71,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 - ))); - } - eprintln!("Build Machine: {}", 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::Timing { - queue_wait, - execution, - } => eprintln!( - "Build queue wait: {:.2}s; execution: {:.2}s", - queue_wait.as_secs_f64(), - execution.as_secs_f64() - ), - 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) @@ -173,6 +140,48 @@ 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); + eprintln!("Build Machine: {}", 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::Timing { + queue_wait, + execution, + } => eprintln!( + "Build queue wait: {:.2}s; execution: {:.2}s", + queue_wait.as_secs_f64(), + execution.as_secs_f64() + ), + 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 5aeec25530..b9b763622a 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)?; @@ -57,12 +64,42 @@ pub(super) fn deploy(root: &ArgMatches) -> Result<(), Error> { let mut options = plan_options(force_recreate, skip_health_monitor); options.selected = selected_attempts(&project, &string_values(matches, "service"))?; let cancellation = crate::cancellation::listen()?; - let (candidate, builds) = - prepare_deploy(matches, &load, project, &resolved, options, &cancellation)?; + let (mut candidate, captured_build) = + prepare_deploy(matches, &load, project, &resolved, options)?; runtime()?.block_on(async { let mut client = crate::cancellation::read(&cancellation, 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 result = build + .execute_remote_images( + &client, + machine.id, + cancellation.clone(), + super::build::progress, + ) + .await; + let cancelled = cancellation.is_cancelled(); + 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(), &cancellation) + .map_err(crate::deploy::DeployError::from)?, + }, + None => Vec::new(), + }; + candidate + .bind_builds(&builds) + .map_err(crate::deploy::DeployError::from)?; deploy_project( &mut client, &candidate, @@ -262,8 +299,7 @@ fn prepare_deploy( mut project: ComposeProject, resolved: &ResolvedProject, options: ployz_core::PlanOptions, - cancellation: &tokio_util::sync::CancellationToken, -) -> Result<(CapturedCompose, Vec), Error> { +) -> Result<(CapturedCompose, Option), Error> { let selected = string_values(matches, "service"); for warning in &project.warnings { eprintln!("WARNING: {warning}"); @@ -277,13 +313,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)?) @@ -297,14 +333,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(), cancellation) - .map_err(crate::deploy::DeployError::from)?, - None => Vec::new(), - }; - Ok((candidate, built)) + Ok((candidate, captured_build)) } fn selected_attempts( @@ -364,3 +393,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..59bab96a1d --- /dev/null +++ b/crates/ployz/src/handlers/deploy_remote_tests.rs @@ -0,0 +1,403 @@ +//! 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(); + *builds.platforms.lock().unwrap() = Some(vec!["linux/arm64".into(), "linux/amd64".into()]); + 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() { + for (architecture, platform) in [("aarch64", "linux/amd64"), ("armv6l", "linux/arm/v7")] { + let (root, service, builds) = fixture(); + let mutations = service.mutating_rpcs(); + *builds.platforms.lock().unwrap() = Some(vec![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] +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!("sha256:{}", "1".repeat(64)), + tags: vec!["registry.invalid/shared:latest".into()], + platforms: vec!["linux/amd64".into()], + location: "unix:///var/run/docker.sock".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, + &[], + &tokio_util::sync::CancellationToken::new(), + ) + .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.repository_reference().unwrap() + && 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()], + &tokio_util::sync::CancellationToken::new(), + ) + .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(); +} + +#[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.platforms.lock().unwrap() = Some(vec![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/handlers/deploy_tests.rs b/crates/ployz/src/handlers/deploy_tests.rs index e8e5c4c649..7a89439903 100644 --- a/crates/ployz/src/handlers/deploy_tests.rs +++ b/crates/ployz/src/handlers/deploy_tests.rs @@ -342,16 +342,8 @@ fn deploy_preparation_captures_resolved_input_and_selection_before_planning() { ..Default::default() }; std::fs::write(&source, yaml.replace("captured", "later-edit")).unwrap(); - let (candidate, builds) = prepare_deploy( - matches, - &load, - project, - &resolved, - options, - &tokio_util::sync::CancellationToken::new(), - ) - .unwrap(); - assert!(builds.is_empty()); + let (candidate, builds) = prepare_deploy(matches, &load, project, &resolved, options).unwrap(); + assert!(builds.is_none()); assert_eq!( std::fs::read_to_string(directory.join("calls")).unwrap(), "x" @@ -417,17 +409,24 @@ 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(), - &tokio_util::sync::CancellationToken::new(), - ) { - 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(), + &tokio_util::sync::CancellationToken::new(), + ) + .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 db0bee22b7..791e861a5b 100644 --- a/crates/ployz/src/image.rs +++ b/crates/ployz/src/image.rs @@ -21,7 +21,10 @@ use crate::{ use self::proxy::{ImageProxy, ProxyMode, detect_mode}; +mod built; mod proxy; +pub use built::push_from_machine; +pub(crate) use built::{platform_compatible, push_from_machine_using_machines, serve_build_image}; #[must_use] pub fn with_default_tag(image: &str) -> String { @@ -38,6 +41,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..9a9aa90a36 --- /dev/null +++ b/crates/ployz/src/image/built.rs @@ -0,0 +1,181 @@ +//! 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], + cancellation: &tokio_util::sync::CancellationToken, +) -> Result, PushError> { + let machines = Cancellation::new(cancellation) + .race(client.machines()) + .await??; + push_from_machine_using_machines(client, image, source, selectors, &machines, cancellation) + .await +} + +pub(crate) async fn push_from_machine_using_machines( + client: &mut Client, + image: &BuiltImage, + source: MachineId, + selectors: &[String], + machines: &[MachineObservation], + cancellation: &tokio_util::sync::CancellationToken, +) -> 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 source = serve_build_image(client, image, source, cancellation).await?; + let reference = image + .repository_reference() + .map_err(|error| PushError::InvalidReference { + reference: image.reference.clone(), + message: error.to_string(), + })?; + let mut cancellation = Cancellation::new(cancellation); + let mut remaining = selection.targets.into_iter(); + while let Some(machine) = remaining.next() { + match pull_on_machine(client, &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, + cancellation: &tokio_util::sync::CancellationToken, +) -> Result { + let mut cancellation = Cancellation::new(cancellation); + let digest = &image.reference; + 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 + && image.platforms.iter().all(|expected| { + stored.platforms.iter().any(|platform| { + platform == expected + || matches!( + (platform.as_str(), expected.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.platforms.join(", "), + }); + } + let opened = cancellation + .race(client.call::(EnsureImageIngestRequest {}, Some(&target))) + .await? + .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/src/lib.rs b/crates/ployz/src/lib.rs index 6189d30977..caf9f1dc6f 100644 --- a/crates/ployz/src/lib.rs +++ b/crates/ployz/src/lib.rs @@ -21,3 +21,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 c03c64509a..5b365ac597 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] @@ -590,6 +591,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') diff --git a/crates/ployz/tests/build/capture.rs b/crates/ployz/tests/build/capture.rs index aba51fb61a..3e1333b3fe 100644 --- a/crates/ployz/tests/build/capture.rs +++ b/crates/ployz/tests/build/capture.rs @@ -576,3 +576,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), &tokio_util::sync::CancellationToken::new()) + .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/railpack.rs b/crates/ployz/tests/build/railpack.rs index 531aace4e3..08ec88b5f0 100644 --- a/crates/ployz/tests/build/railpack.rs +++ b/crates/ployz/tests/build/railpack.rs @@ -432,6 +432,7 @@ fn railpack_cancellation_stops_work_and_releases_private_inputs_and_admission() refresh_cache: false, }]; let request = Request { + image_contexts: &BTreeMap::new(), compose_file: &root.join("compose.yaml"), working_dir: &root, environment: &BTreeMap::from([("PATH".into(), std::env::var("PATH").unwrap())]), @@ -560,6 +561,7 @@ fn multi_platform_failures_preserve_output_stage_and_completed_image_evidence() ..HostPolicy::default() }; let request = Request { + image_contexts: &BTreeMap::new(), compose_file: &capture.join("compose.yaml"), working_dir: &capture, environment: &environment, diff --git a/crates/ployz/tests/build_layer3/remote.rs b/crates/ployz/tests/build_layer3/remote.rs index 88270173e6..25a05a89d5 100644 --- a/crates/ployz/tests/build_layer3/remote.rs +++ b/crates/ployz/tests/build_layer3/remote.rs @@ -78,7 +78,7 @@ async fn remote_dockerfile_runs_on_selected_machine_and_bounds_abandoned_attempt assert!(stdout.contains(selected.as_str()), "{stdout}"); let exact = stdout .split_whitespace() - .find(|word| word.contains("@sha256:")) + .find(|word| word.starts_with("sha256:")) .expect("remote exact image identity"); assert_eq!( cluster @@ -224,6 +224,8 @@ 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(), platforms: Vec::new(), @@ -365,7 +367,7 @@ async fn queue_upload_timeout_and_daemon_restart_discard_waiters_before_a_safe_b let stdout = String::from_utf8(output.stdout).unwrap(); let exact = stdout .split_whitespace() - .find(|word| word.contains("@sha256:")) + .find(|word| word.starts_with("sha256:")) .unwrap(); assert_eq!( cluster @@ -375,3 +377,176 @@ async fn queue_upload_timeout_and_daemon_restart_discard_waiters_before_a_safe_b ); fs::remove_dir_all(root).unwrap(); } + +#[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()); + assert_temporary_tags_released(&cluster).await; + 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()], + &CancellationToken::new(), + ) + .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" + ); + 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/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..c6fd094877 --- /dev/null +++ b/crates/ployz/tests/support/build.rs @@ -0,0 +1,163 @@ +//! 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 platforms: 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, + active_timeout: ployz_build::EXECUTION_TIMEOUT, + }) + .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!( + "sha256:{}", + (offset + index + 1).to_string().repeat(64) + ), + tags: vec!["registry.invalid/shared:latest".into()], + platforms: self + .platforms + .lock() + .unwrap() + .clone() + .unwrap_or_else(|| vec!["linux/amd64".into()]), + location: "unix:///var/run/docker.sock".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.clone(), + repo_tags: image.tags.clone(), + created: 0, + size: 1, + containers: 0, + platforms: image.platforms.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/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 e745ebaef7..dba2ce00dc 100644 --- a/crates/ployzd/src/build.rs +++ b/crates/ployzd/src/build.rs @@ -9,11 +9,21 @@ use ployz_build::{ remote::{self, Definition, Event, Input, Outcome}, }; 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, @@ -21,7 +31,9 @@ pub(crate) fn start( ) -> RpcStream { let (events, receiver) = mpsc::channel(8); tokio::spawn(async move { - let outcome = attempt(machine_id, requests, &events, runner).await; + let shutdown = runner.shutdown.clone(); + let state = Arc::new(AttemptState::default()); + let outcome = attempt(machine_id, requests, &events, runner, state.clone()).await; let message = "Build terminal report exceeds the response size limit"; let fallback = match &outcome { Outcome::Unknown { stage, .. } => Outcome::Unknown { @@ -38,6 +50,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() + { + tokio::select! { () = events.closed() => {}, () = shutdown.cancelled() => {} } + } + let retained = state.retained.lock().expect("Build retention lock").take(); + let _ = tokio::task::spawn_blocking(move || drop(retained)).await; }); ReceiverStream::new(receiver) } @@ -46,7 +69,8 @@ async fn attempt( machine_id: MachineId, mut requests: impl Stream> + Send + Unpin + 'static, events: &mpsc::Sender>, - runner: std::sync::Arc, + runner: Arc, + state: Arc, ) -> Outcome { let start = tokio::time::timeout(Duration::from_secs(10), requests.next()).await; let definition = match start { @@ -56,7 +80,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", @@ -102,10 +129,9 @@ async fn attempt( Ok(Err(error)) => return failure(Stage::Admission, error).with_work(work), 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; @@ -131,7 +157,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, () = runner.shutdown.cancelled() => "Build daemon stopped".to_owned(), () = events.closed() => "Build client disconnected".to_owned(), @@ -164,7 +190,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( @@ -192,12 +218,12 @@ async fn pump( fn receive_and_execute( machine_id: MachineId, - definition: Definition, + mut definition: Definition, mut source: mpsc::Receiver, admission: Admission, policy: HostPolicy, events: &mpsc::Sender>, - evidence: &std::sync::Mutex, + state: &AttemptState, ) -> Outcome { let cancellation = admission.cancellation(); let deadline = std::time::Instant::now() + admission.remaining(); @@ -225,7 +251,8 @@ fn receive_and_execute( } } let progress = |progress| { - evidence + state + .evidence .lock() .expect("Build evidence lock") .observe(&progress); @@ -248,12 +275,36 @@ fn receive_and_execute( } }; progress(Progress::Stage(Stage::Upload)); - match upload.execute(&definition, 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 }, - }, + // 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); + } + let result = upload.execute(&definition, 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 91aa4c72df..c2780363c3 100644 --- a/crates/ployzd/src/build/tests.rs +++ b/crates/ployzd/src/build/tests.rs @@ -17,6 +17,7 @@ use tokio_stream::wrappers::TcpListenerStream; use tonic::transport::Server; struct Fixture { + shutdown: tokio_util::sync::CancellationToken, root: PathBuf, policy: HostPolicy, cluster: tokio::task::JoinHandle<()>, @@ -71,7 +72,8 @@ impl Fixture { ..Default::default() }; update(&mut policy); - service.builds = Runner::new(policy.clone(), Default::default()).unwrap(); + let shutdown = tokio_util::sync::CancellationToken::new(); + service.builds = Runner::new(policy.clone(), shutdown.clone()).unwrap(); write_docker(&policy.docker, &root); let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let address = format!("http://{}", listener.local_addr().unwrap()); @@ -80,6 +82,7 @@ impl Fixture { Server::builder().serve_with_incoming(api, TcpListenerStream::new(listener)), ); Self { + shutdown, root, policy, cluster, @@ -99,6 +102,8 @@ 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(), platforms: Vec::new(), @@ -318,6 +323,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 @@ -378,6 +384,8 @@ 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(), platforms: Vec::new(), @@ -467,7 +475,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 + }) ); } } @@ -680,3 +693,62 @@ 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() { + for stopping in [false, true] { + 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(); + // Holding completed images must not occupy the Machine's active Build slot. + let next = tokio::time::timeout( + Duration::from_secs(5), + fixture.capture().execute_remote_images( + &client, + fixture.machine.id, + tokio_util::sync::CancellationToken::new(), + |_| {}, + ), + ) + .await + .expect("image retention held queue admission") + .unwrap(); + assert!(!fixture.root.join("released-tags").exists()); + drop(next); + if stopping { + fixture.shutdown.cancel(); + } else { + drop(images); + } + tokio::time::timeout(Duration::from_secs(3), async { + while fs::read_to_string(fixture.root.join("released-tags")) + .unwrap_or_default() + .lines() + .count() + < 2 + { + 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/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(), 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| {