diff --git a/README.md b/README.md index 70828f5aad..a74344c630 100644 --- a/README.md +++ b/README.md @@ -38,6 +38,23 @@ Railpack refuses `--check` and unsupported frontend settings by name. On this pinned frontend, `--no-cache` and `--pull` force a cold build by clearing the exclusive Ployz builder cache; unrelated Docker builder caches are untouched. +Remote Builds (`ployz build --remote=`) use one active +slot per Machine and a FIFO of eight waiting attempts. Source and secrets stay +on the client until admission. Configure the daemon environment and restart it: + +| Setting | Default | Accepted values | +| --- | --- | --- | +| `PLOYZ_BUILD_QUEUE_CAPACITY` | `8` | `0`–`1024` waiting attempts | +| `PLOYZ_BUILD_QUEUE_TIMEOUT_SECONDS` | `600` | `1`–`86400` seconds | +| `PLOYZ_BUILD_ACTIVE_TIMEOUT_SECONDS` | `1800` | `1`–`86400` seconds | + +The active budget includes upload, preparation, execution, import and cleanup. +Cancellation and disconnect remove waiters; daemon restart discards the queue. +Unconfirmed termination keeps builder ownership quarantined, including across +restart. Bounded abandoned-builder teardown does not clear that uncertainty: +an operator must confirm the builder and its host processes have stopped before +clearing the lock marker named in the error. Work is never replayed. + `ployz build --remote=` runs Dockerfile or native-platform Railpack Builds on the selected Machine. That Machine supplies the build resource policy; source uploads and build requests cannot change it. diff --git a/crates/ployz-build/src/builder.rs b/crates/ployz-build/src/builder.rs index 58a9fa0726..ddf7e3662e 100644 --- a/crates/ployz-build/src/builder.rs +++ b/crates/ployz-build/src/builder.rs @@ -248,10 +248,14 @@ fn remove(docker: &Docker<'_>, name: &str) -> Result<(), BuildError> { /// Serializes local attempts sharing one builder and its retained cache. /// /// Held for the whole attempt. Uncertain termination quarantines the retained state. +#[derive(Clone)] pub(crate) struct Lock { - file: fs::File, + file: std::sync::Arc, + pub(crate) directory: PathBuf, } +struct LockedFile(fs::File); + impl Lock { /// Wait for exclusive use of this user's builder. /// @@ -274,6 +278,52 @@ impl Lock { } pub(crate) fn try_acquire_in(directory: &std::path::Path) -> Result { + let lock = Self::open_locked(directory)?; + if lock.file.0.metadata().map_err(lock_error)?.len() != 0 { + return Err(lock.uncertain()); + } + Ok(lock) + } + + fn uncertain(&self) -> BuildError { + BuildError::UncertainTermination(format!( + "retained builder state is quarantined; confirm builder {} and its host processes have stopped before clearing {}", + builder_name(), + self.directory + .join(format!("{}.lock", builder_name())) + .display() + )) + } + + /// One bounded teardown of an abandoned builder on daemon startup. The + /// marker remains: removing a container cannot prove an orphaned host + /// process stopped, so cleanup alone must never authorize conflicting work. + pub(crate) fn cleanup_abandoned(policy: &crate::HostPolicy) -> Result<(), BuildError> { + let lock = Self::open_locked(&policy.state_directory)?; + if lock.file.0.metadata().map_err(lock_error)?.len() == 0 { + return crate::upload::remove_abandoned(&lock.directory.join("build-upload")).map_err( + |error| BuildError::Prerequisite(format!("remove abandoned Build upload: {error}")), + ); + } + let environment = crate::upload::environment(&lock.directory.join("build-upload")); + let docker = Docker { + program: &policy.docker, + environment: &environment, + working_dir: &lock.directory, + deadline: crate::Deadline::starting_now(crate::CLEANUP_TIMEOUT), + cancellation: None, + progress: None, + }; + if let Err(error) = remove(&docker, &builder_name()) { + return Err(BuildError::UncertainTermination(format!( + "{}; abandoned builder cleanup failed: {error}", + lock.uncertain() + ))); + } + Err(lock.uncertain()) + } + + fn open_locked(directory: &std::path::Path) -> Result { use rustix::fs::{FlockOperation, flock}; use std::os::unix::fs::MetadataExt as _; match fs::DirBuilder::new().mode(0o700).create(directory) { @@ -305,36 +355,32 @@ impl Lock { Err(rustix::io::Errno::WOULDBLOCK) => return Err(BuildError::Busy), Err(error) => return Err(lock_error(error.into())), } - if file.metadata().map_err(lock_error)?.len() != 0 { - return Err(BuildError::UncertainTermination(format!( - "retained builder state is quarantined; confirm builder {} has stopped before clearing {}", - builder_name(), - path.display() - ))); - } - Ok(Self { file }) + Ok(Self { + file: std::sync::Arc::new(LockedFile(file)), + directory: directory.to_owned(), + }) } - fn clear(&mut self) -> Result<(), BuildError> { - self.file.set_len(0).map_err(lock_error)?; - self.file.sync_all().map_err(lock_error) + pub(crate) fn clear(&mut self) -> Result<(), BuildError> { + self.file.0.set_len(0).map_err(lock_error)?; + self.file.0.sync_all().map_err(lock_error) } /// Persist uncertainty across process exit before releasing the OS lock. pub(crate) fn quarantine(&mut self) -> Result<(), BuildError> { use std::io::Write as _; - self.file + (&self.file.0) .write_all(b"termination unconfirmed\n") .map_err(lock_error)?; - self.file.sync_all().map_err(lock_error) + self.file.0.sync_all().map_err(lock_error) } } -impl Drop for Lock { +impl Drop for LockedFile { fn drop(&mut self) { // Another thread may have forked a child that briefly inherited this // CLOEXEC descriptor. Release ownership now, without waiting for exec. - let _ = rustix::fs::flock(&self.file, rustix::fs::FlockOperation::Unlock); + let _ = rustix::fs::flock(&self.0, rustix::fs::FlockOperation::Unlock); } } @@ -438,8 +484,9 @@ esac let mut builder = Builder::acquire(&docker, lock, &crate::policy::Resources::default()).unwrap(); // A read-only descriptor deterministically makes marker truncation fail. - builder.lock.as_mut().unwrap().file = - fs::File::open(directory.join(format!("{}.lock", builder_name()))).unwrap(); + builder.lock.as_mut().unwrap().file = std::sync::Arc::new(LockedFile( + fs::File::open(directory.join(format!("{}.lock", builder_name()))).unwrap(), + )); let result = if prior_failure { Err(BuildError::Result("earlier build failure".into()).at(crate::Stage::Building)) } else { diff --git a/crates/ployz-build/src/execution.rs b/crates/ployz-build/src/execution.rs index e1bc568f1f..f8119c42a5 100644 --- a/crates/ployz-build/src/execution.rs +++ b/crates/ployz-build/src/execution.rs @@ -15,6 +15,7 @@ use std::{ #[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)] pub enum Stage { Admission, + Queued, Upload, Preparation, Building, @@ -27,6 +28,11 @@ pub enum Stage { pub enum Progress { Stage(Stage), Output(Vec), + /// Client-observed waiting and admitted execution, measured separately. + Timing { + queue_wait: Duration, + execution: Duration, + }, /// Proven per-target work, retained even when a later step fails. Target { name: String, @@ -92,6 +98,10 @@ pub struct HostPolicy { pub docker: PathBuf, /// Total active budget beginning at admission. pub active_timeout: Duration, + /// Maximum waiting attempts, excluding the active attempt. + pub queue_capacity: usize, + /// Waiting budget; execution starts a separate clock after admission. + pub queue_timeout: Duration, } impl Default for HostPolicy { fn default() -> Self { @@ -102,10 +112,65 @@ impl Default for HostPolicy { .join(".ployz/build.yaml"), docker: "docker".into(), active_timeout: EXECUTION_TIMEOUT, + queue_capacity: 8, + queue_timeout: Duration::from_secs(600), } } } +impl HostPolicy { + /// Read Machine-local settings once at daemon startup. + /// # Errors + /// Rejects non-numeric, zero timeout, and excessive settings. + pub fn from_environment() -> Result { + Self::from_settings(|name| { + std::env::var_os(name).map(|value| value.to_string_lossy().into_owned()) + }) + } + + fn from_settings(get: impl Fn(&str) -> Option) -> Result { + let mut policy = Self::default(); + if let Some(raw) = get("PLOYZ_BUILD_QUEUE_CAPACITY") { + policy.queue_capacity = raw.parse().map_err(|_| { + BuildError::Prerequisite("invalid PLOYZ_BUILD_QUEUE_CAPACITY".into()) + })?; + } + for (name, value) in [ + ( + "PLOYZ_BUILD_QUEUE_TIMEOUT_SECONDS", + &mut policy.queue_timeout, + ), + ( + "PLOYZ_BUILD_ACTIVE_TIMEOUT_SECONDS", + &mut policy.active_timeout, + ), + ] { + if let Some(raw) = get(name) { + *value = Duration::from_secs( + raw.parse() + .map_err(|_| BuildError::Prerequisite(format!("invalid {name}")))?, + ); + } + } + policy.validate()?; + Ok(policy) + } + + /// Validate before allocating queue state or computing deadlines. + /// # Errors + /// Capacity is 0–1024; timeouts must be positive and at most 24 hours. + pub fn validate(&self) -> Result<(), BuildError> { + if self.queue_capacity > 1024 + || [self.queue_timeout, self.active_timeout] + .iter() + .any(|timeout| timeout.is_zero() || *timeout > Duration::from_secs(86400)) + { + return Err(BuildError::Prerequisite("Build policy requires queue capacity 0–1024 and positive timeouts no greater than 86400 seconds".into())); + } + Ok(()) + } +} + /// Exclusive retained-builder ownership acquired before accepting source. /// Dropping unused admission is safe: no execution has started. pub struct Admission { @@ -115,10 +180,28 @@ pub struct Admission { pub(crate) resources: crate::policy::Resources, } impl Admission { + /// Attempt abandoned-resource teardown once after daemon restart. Never + /// clears uncertainty about orphaned host processes or replays a Build. + /// # Errors + /// Busy or uncertain ownership remains unavailable. + pub fn cleanup_abandoned(policy: &HostPolicy) -> Result<(), BuildError> { + policy.validate()?; + Lock::cleanup_abandoned(policy) + } + + /// Receive into Machine-owned staging under this admission. Interrupted + /// uploads from a previous daemon are removed before this path is reused. + /// # Errors + /// Refuses unsafe ownership or failure to remove/create protected staging. + pub fn upload(self) -> Result { + crate::upload::AdmittedUpload::new(self) + } + /// Acquire under host policy, including the deadline that starts before upload. /// # Errors /// Refuses busy or quarantined state and reports filesystem failures. pub fn try_acquire_with(policy: &HostPolicy) -> Result { + policy.validate()?; let resources = crate::policy::Resources::load(&policy.configuration_file)?; Ok(Self { resources, @@ -154,8 +237,165 @@ impl Admission { return Err(BuildError::Cancelled); } if self.remaining().is_zero() { - return Err(BuildError::TimedOut(EXECUTION_TIMEOUT.as_secs())); + return Err(BuildError::TimedOut(self.deadline.budget.as_secs())); } Ok(()) } } + +#[cfg(test)] +mod tests { + use super::*; + + #[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(); + let policy = HostPolicy { + state_directory: root.clone(), + docker: root.join("docker"), + ..Default::default() + }; + crate::tests::executable( + &policy.docker, + &format!( + "#!/bin/sh\n[ \"$1\" = --ready ] && exit 0\nprintf cleaned > '{}/cleaned'\n", + root.display() + ), + ); + std::fs::create_dir(root.join("build-upload")).unwrap(); + std::fs::write(root.join("build-upload/abandoned"), "private old capture").unwrap(); + Admission::cleanup_abandoned(&policy).unwrap(); + assert!( + !root.join("build-upload").exists(), + "startup left abandoned inputs" + ); + let admission = Admission::try_acquire_with(&policy).unwrap(); + let upload = admission.upload().unwrap(); + assert!(!root.join("build-upload/abandoned").exists()); + assert!(matches!( + Admission::try_acquire_with(&policy), + Err(BuildError::Busy) + )); + drop(upload); + let mut lock = Lock::try_acquire_in(&root).unwrap(); + lock.quarantine().unwrap(); + drop(lock); + assert!( + Admission::cleanup_abandoned(&policy) + .unwrap_err() + .is_unknown() + ); + assert!( + root.join("cleaned").exists(), + "cleanup skipped absent upload staging" + ); + std::fs::remove_file(root.join("cleaned")).unwrap(); + std::fs::create_dir(root.join("build-upload")).unwrap(); + std::fs::write(root.join("build-upload/uncertain"), "still in use").unwrap(); + assert!( + Admission::cleanup_abandoned(&policy) + .unwrap_err() + .is_unknown() + ); + assert!( + root.join("cleaned").exists(), + "bounded abandoned cleanup was not attempted" + ); + assert!( + root.join("build-upload/uncertain").exists(), + "unknown work lost private inputs" + ); + assert!( + Admission::try_acquire_with(&policy) + .err() + .unwrap() + .is_unknown() + ); + std::fs::remove_dir_all(root).unwrap(); + } + + #[test] + 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(); + let policy = HostPolicy { + state_directory: root.clone(), + active_timeout: Duration::from_secs(5), + ..Default::default() + }; + let mut admission = Admission::try_acquire_with(&policy).unwrap(); + admission.deadline.expires = std::time::Instant::now(); + assert!(matches!(admission.check(), Err(BuildError::TimedOut(5)))); + let environment = std::collections::BTreeMap::new(); + let docker = crate::Docker { + program: &policy.docker, + environment: &environment, + working_dir: &root, + deadline: admission.deadline, + cancellation: None, + progress: None, + }; + assert!(matches!( + docker.run("build", &["build"], crate::Streams::Captured), + Err(BuildError::TimedOut(5)) + )); + drop(admission); + std::fs::remove_dir_all(root).unwrap(); + } + + #[test] + fn cancelled_admission_cannot_delete_or_create_upload_staging() { + 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::write(root.join("build-upload/existing"), "capture").unwrap(); + let policy = HostPolicy { + state_directory: root.clone(), + ..Default::default() + }; + let admission = Admission::try_acquire_with(&policy).unwrap(); + admission.cancellation().cancel(); + assert!(admission.upload().is_err()); + assert!(root.join("build-upload/existing").exists()); + assert!(Admission::try_acquire_with(&policy).is_ok()); + std::fs::remove_dir_all(root).unwrap(); + } + + #[test] + fn machine_policy_defaults_and_invalid_input_are_bounded() { + let policy = HostPolicy::from_settings(|_| None).unwrap(); + assert_eq!(policy.queue_capacity, 8); + assert_eq!(policy.queue_timeout, Duration::from_secs(600)); + assert_eq!(policy.active_timeout, Duration::from_secs(1800)); + for (name, values) in [ + ("PLOYZ_BUILD_QUEUE_CAPACITY", vec!["-1", "1025", "abc"]), + ( + "PLOYZ_BUILD_QUEUE_TIMEOUT_SECONDS", + vec!["0", "86401", "18446744073709551615"], + ), + ( + "PLOYZ_BUILD_ACTIVE_TIMEOUT_SECONDS", + vec!["0", "86401", "invalid"], + ), + ] { + for value in values { + assert!( + HostPolicy::from_settings(|key| (key == name).then(|| value.into())).is_err(), + "{name}={value}" + ); + } + } + let policy = HostPolicy::from_settings(|key| match key { + "PLOYZ_BUILD_QUEUE_CAPACITY" => Some("0".into()), + "PLOYZ_BUILD_QUEUE_TIMEOUT_SECONDS" => Some("12".into()), + "PLOYZ_BUILD_ACTIVE_TIMEOUT_SECONDS" => Some("34".into()), + _ => None, + }) + .unwrap(); + assert_eq!(policy.queue_capacity, 0); + assert_eq!(policy.queue_timeout, Duration::from_secs(12)); + assert_eq!(policy.active_timeout, Duration::from_secs(34)); + } +} diff --git a/crates/ployz-build/src/lib.rs b/crates/ployz-build/src/lib.rs index 8a9a4a2264..01a62822ee 100644 --- a/crates/ployz-build/src/lib.rs +++ b/crates/ployz-build/src/lib.rs @@ -231,7 +231,7 @@ pub fn execute(request: &Request<'_>) -> Result, BuildError> { /// resources cannot be confirmed stopped. Uncertainty blocks competing work. pub fn execute_admitted( request: &Request<'_>, - admission: Admission, + mut admission: Admission, progress: &(dyn Fn(Progress) + Sync), ) -> Result, BuildError> { let planned = plan(request.targets)?; @@ -249,12 +249,24 @@ pub fn execute_admitted( }; let metadata = request.working_dir.join("build-metadata.json"); progress(Progress::Stage(Stage::Preparation)); - docker - .require_local() + // Preflight processes also use private inputs. Persist ownership before + // spawning them, including across a daemon crash or uncertain termination. + admission + .lock + .quarantine() .map_err(|error| error.at(Stage::Preparation))?; docker - .run("check Buildx", &["buildx", "version"], Streams::Captured) - .map_err(|error| error.at(Stage::Preparation))?; + .require_local() + .and_then(|()| docker.run("check Buildx", &["buildx", "version"], Streams::Captured)) + .map_err(|error| { + let error = error.at(Stage::Preparation); + if !error.is_unknown() + && let Err(cleanup) = admission.lock.clear() + { + return error.with_later_failure(cleanup); + } + error + })?; let builder = Builder::acquire(&docker, admission.lock, &admission.resources) .map_err(|error| error.at(Stage::Preparation))?; let result = (|| { @@ -708,7 +720,7 @@ impl<'a> Docker<'a> { return Err(BuildError::Cancelled); } if self.deadline.remaining().is_zero() { - return Err(BuildError::TimedOut(EXECUTION_TIMEOUT.as_secs())); + return Err(BuildError::TimedOut(self.deadline.budget.as_secs())); } let mut command = Command::new(self.program); command diff --git a/crates/ployz-build/src/remote.rs b/crates/ployz-build/src/remote.rs index f4d670c9cd..7a6f5580f2 100644 --- a/crates/ployz-build/src/remote.rs +++ b/crates/ployz-build/src/remote.rs @@ -6,7 +6,7 @@ use ployz_core::{MachineId, OpaquePayload}; use serde::{Deserialize, Serialize, de::DeserializeOwned}; pub use crate::received_recipe::{validate_capture, validate_remote_context}; -pub use crate::upload::{Upload, upload}; +pub use crate::upload::{AdmittedUpload, Upload, upload}; /// Invalid or unavailable captured input at the transport trust boundary. #[derive(Debug, thiserror::Error)] @@ -69,7 +69,10 @@ pub enum Input { /// Admission, observed progress, or the terminal host report. #[derive(Clone, Debug, Serialize, Deserialize)] pub enum Event { - Admitted { machine_id: MachineId }, + Admitted { + machine_id: MachineId, + active_timeout: std::time::Duration, + }, Progress(Progress), Finished(Outcome), } diff --git a/crates/ployz-build/src/tests.rs b/crates/ployz-build/src/tests.rs index 62c462af73..e8d0b66201 100644 --- a/crates/ployz-build/src/tests.rs +++ b/crates/ployz-build/src/tests.rs @@ -211,6 +211,7 @@ exit 0 state_directory: directory.clone(), active_timeout: EXECUTION_TIMEOUT, configuration_file: directory.join("build.yaml"), + ..Default::default() }) .unwrap(); let cancellation = admission.cancellation(); @@ -359,3 +360,61 @@ fn a_command_that_outlasts_its_budget_is_terminated() { )); assert!(waited.elapsed() < Duration::from_secs(5)); } + +#[test] +fn docker_preflight_holds_quarantine_and_clears_confirmed_failures() { + let directory = std::env::temp_dir().join(format!("ployz-preflight-{}", uuid::Uuid::new_v4())); + use std::os::unix::fs::DirBuilderExt as _; + std::fs::DirBuilder::new() + .mode(0o700) + .create(&directory) + .unwrap(); + let program = directory.join("docker"); + let policy = HostPolicy { + state_directory: directory.clone(), + configuration_file: directory.join("build.yaml"), + ..Default::default() + }; + let targets = [target("api", None)]; + let environment = BTreeMap::new(); + let request = Request { + railpack: &[], + compose_file: &directory.join("compose.json"), + working_dir: &directory, + environment: &environment, + docker: Some(&program), + targets: &targets, + build_args: &[], + output: Output::Load, + no_cache: false, + pull: false, + }; + for stop_at in ["context show", "buildx version"] { + executable( + &program, + &format!( + r#"#!/bin/sh +[ "$1" = --ready ] && exit 0 +[ -s '{}.lock' ] || {{ echo missing-quarantine >&2; exit 1; }} +if [ "$*" = '{stop_at}' ]; then + echo observed-quarantine >&2 + exit 1 +fi +[ "$*" = 'context show' ] && echo default +exit 0 +"#, + builder_name() + ), + ); + let admission = Admission::try_acquire_with(&policy).unwrap(); + let error = execute_admitted(&request, admission, &|_| {}).unwrap_err(); + assert!( + error.to_string().contains("observed-quarantine"), + "{stop_at}: {error}" + ); + assert_eq!(error.stage(), Stage::Preparation); + assert!(!error.is_unknown()); + assert!(Admission::try_acquire_with(&policy).is_ok()); + } + std::fs::remove_dir_all(directory).unwrap(); +} diff --git a/crates/ployz-build/src/upload.rs b/crates/ployz-build/src/upload.rs index bca79d993d..36b5ced5c7 100644 --- a/crates/ployz-build/src/upload.rs +++ b/crates/ployz-build/src/upload.rs @@ -46,6 +46,49 @@ enum UploadState { /// A finished capture; only this type can execute. pub struct CompletedUpload(Upload); +/// A receiver that owns its admission until source cleanup finishes. +/// Consuming Admission prevents two receivers from sharing live staging. +pub struct AdmittedUpload { + upload: Upload, + admission: Admission, +} + +impl AdmittedUpload { + pub(crate) fn new(admission: Admission) -> Result { + admission + .check() + .map_err(|error| InputError::from(error.to_string()))?; + let upload = Upload::owned(admission.lock.directory.join("build-upload"))?; + Ok(Self { upload, admission }) + } + + /// Validate the next frame within the admitted execution budget. + /// # Errors + /// Rejects cancellation, expiry, and malformed upload frames. + pub fn accept(&mut self, frame: Input) -> Result<(), InputError> { + self.admission + .check() + .map_err(|error| InputError::from(error.to_string()))?; + self.upload.accept(frame) + } + + /// Finish receiving and execute while retaining ownership through cleanup. + /// # Errors + /// Rejects incomplete captures or host execution failure. + pub fn execute( + self, + definition: &Definition, + docker: Option<&Path>, + progress: &(dyn Fn(Progress) + Sync), + ) -> Result, BuildError> { + let _ownership = self.admission.lock.clone(); + self.upload + .complete() + .map_err(|error| BuildError::Request(error.to_string()).at(crate::Stage::Upload))? + .execute(definition, self.admission, docker, progress) + } +} + impl Upload { /// Create private staging for one bounded upload. /// # Errors @@ -53,6 +96,17 @@ impl Upload { pub fn new() -> Result { let root = std::env::temp_dir().join(format!("ployz-received-build-{}", uuid::Uuid::new_v4())); + Self::create(root) + } + + pub(crate) fn owned(root: PathBuf) -> Result { + // Called only with retained-builder ownership. An execution marker + // prevents reaching this point while abandoned work is uncertain. + remove_abandoned(&root).map_err(io_error)?; + Self::create(root) + } + + fn create(root: PathBuf) -> Result { fs::DirBuilder::new() .mode(0o700) .create(&root) @@ -235,24 +289,7 @@ impl CompletedUpload { fs::set_permissions(upload.root.join(path), fs::Permissions::from_mode(*mode)) .map_err(|error| BuildError::Request(io_error(error).to_string()))?; } - let environment = BTreeMap::from([ - ( - "PATH".into(), - std::env::var("PATH").unwrap_or_else(|_| "/usr/local/bin:/usr/bin:/bin".into()), - ), - ( - "HOME".into(), - upload.root.join("private").to_string_lossy().into_owned(), - ), - ( - "DOCKER_CONFIG".into(), - upload - .root - .join("private/docker") - .to_string_lossy() - .into_owned(), - ), - ]); + let environment = environment(&upload.root); let result = crate::execute_admitted( &Request { railpack: &railpack, @@ -285,6 +322,15 @@ impl Drop for Upload { } } +/// Caller must hold builder ownership and verify that no execution is uncertain. +pub(crate) fn remove_abandoned(root: &Path) -> io::Result<()> { + if root.try_exists()? { + make_removable(root)?; + fs::remove_dir_all(root)?; + } + Ok(()) +} + fn make_removable(path: &Path) -> io::Result<()> { if fs::symlink_metadata(path)?.is_dir() { fs::set_permissions(path, fs::Permissions::from_mode(0o700))?; @@ -413,6 +459,25 @@ pub fn upload( send(Input::Finish) } +/// The same private Docker configuration is used for execution and bounded +/// abandoned-builder cleanup; no client paths or ambient credentials enter it. +pub(crate) fn environment(root: &Path) -> BTreeMap { + BTreeMap::from([ + ( + "PATH".into(), + std::env::var("PATH").unwrap_or_else(|_| "/usr/local/bin:/usr/bin:/bin".into()), + ), + ( + "HOME".into(), + root.join("private").to_string_lossy().into_owned(), + ), + ( + "DOCKER_CONFIG".into(), + root.join("private/docker").to_string_lossy().into_owned(), + ), + ]) +} + #[cfg(test)] mod tests { use super::*; diff --git a/crates/ployz-build/tests/host_policy.rs b/crates/ployz-build/tests/host_policy.rs index 3c20d72df5..59604c2e31 100644 --- a/crates/ployz-build/tests/host_policy.rs +++ b/crates/ployz-build/tests/host_policy.rs @@ -33,6 +33,7 @@ impl Host { state_directory: root, docker, active_timeout: Duration::from_secs(10), + ..Default::default() }, } } diff --git a/crates/ployz/src/compose/remote_build.rs b/crates/ployz/src/compose/remote_build.rs index e665ac7a19..29251ae250 100644 --- a/crates/ployz/src/compose/remote_build.rs +++ b/crates/ployz/src/compose/remote_build.rs @@ -20,6 +20,49 @@ pub(super) async fn execute( machine_id: MachineId, cancellation: CancellationToken, progress: impl Fn(Progress), +) -> Outcome { + let mut queued = None; + let mut executing = None; + let outcome = execute_attempt( + inputs, + definition, + client, + machine_id, + cancellation, + |event| { + match &event { + Progress::Stage(Stage::Queued) => { + queued.get_or_insert_with(tokio::time::Instant::now); + } + Progress::Stage(Stage::Upload) => { + executing.get_or_insert_with(tokio::time::Instant::now); + } + Progress::Stage(_) + | Progress::Output(_) + | Progress::Timing { .. } + | Progress::Target { .. } => {} + } + progress(event); + }, + ) + .await; + let finished = tokio::time::Instant::now(); + progress(Progress::Timing { + queue_wait: queued.map_or(Duration::ZERO, |start| { + executing.unwrap_or(finished) - start + }), + execution: executing.map_or(Duration::ZERO, |start| finished - start), + }); + outcome +} + +async fn execute_attempt( + inputs: BuildInputs, + definition: Definition, + client: &Client, + 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) { @@ -54,28 +97,35 @@ pub(super) async fn execute( ); } }; - match tokio::time::timeout(Duration::from_secs(10), responses.message()).await { - Ok(Ok(Some(payload))) => match remote::decode::(&payload) { - Ok(Event::Admitted { - machine_id: admitted, - }) if admitted == machine_id => {} - Ok(Event::Finished(outcome @ (Outcome::Failed { .. } | Outcome::Unknown { .. }))) => { - return outcome.with_work(evidence); - } - _ => { - return failed( - Stage::Admission, - "invalid Build admission response; source was not uploaded", - ); + 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); } - }, - _ => { - return failed( - Stage::Admission, - "Build admission was not observed; source was not uploaded", - ); + 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(); @@ -101,8 +151,7 @@ pub(super) async fn execute( let mut uploaded = false; let mut cancelling = false; let mut stage = Stage::Upload; - let mut deadline = - tokio::time::Instant::now() + ployz_build::EXECUTION_TIMEOUT + Duration::from_secs(70); + let mut deadline = tokio::time::Instant::now() + active_timeout + Duration::from_secs(70); loop { tokio::select! { result = responses.message() => match result { diff --git a/crates/ployz/src/handlers/build.rs b/crates/ployz/src/handlers/build.rs index c0206fa2d4..c20641afeb 100644 --- a/crates/ployz/src/handlers/build.rs +++ b/crates/ployz/src/handlers/build.rs @@ -87,6 +87,7 @@ pub(super) fn run(matches: &ArgMatches) -> Result<(), Error> { machine.id ))); } + eprintln!("Build Machine: {}", machine.id); let captured = capture_build(&plan, &options, &mut project)?; let outcome = captured .execute_remote(&client, machine.id, cancellation.clone(), |event| { @@ -94,6 +95,14 @@ pub(super) fn run(matches: &ArgMatches) -> Result<(), Error> { 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); } diff --git a/crates/ployz/tests/build_layer3/remote.rs b/crates/ployz/tests/build_layer3/remote.rs index cea7a6bb1e..67e2d0c3df 100644 --- a/crates/ployz/tests/build_layer3/remote.rs +++ b/crates/ployz/tests/build_layer3/remote.rs @@ -95,12 +95,21 @@ async fn remote_dockerfile_runs_on_selected_machine_and_bounds_abandoned_attempt let (first, mut response) = request(&address, selected).await; assert!( - matches!(event(&mut response).await, Event::Admitted { machine_id } if machine_id == selected) - ); - let (_competitor, mut rejected) = request(&address, selected).await; - assert!( - matches!(terminal(&mut rejected).await, Outcome::Failed { stage: Stage::Admission, message, .. } if message.contains("busy")) + matches!(event(&mut response).await, Event::Admitted { machine_id, .. } if machine_id == selected) ); + let (competitor, mut queued) = request(&address, selected).await; + assert!(matches!( + event(&mut queued).await, + Event::Progress(Progress::Stage(Stage::Queued)) + )); + drop(competitor); + assert!(matches!( + terminal(&mut queued).await, + Outcome::Failed { + stage: Stage::Queued, + .. + } + )); for frame in [ Input::Entry { path: b"source".to_vec(), @@ -246,3 +255,123 @@ async fn terminal(response: &mut tonic::Streaming) -> Outcome { } } } + +#[tokio::test] +#[ignore = "informing: requires the privileged Ployz testkit image with Buildx"] +async fn queue_upload_timeout_and_daemon_restart_discard_waiters_before_a_safe_build() { + let mut plan = ClusterPlan::new(&format!("l3-build-807-{}", std::process::id()), 1).unwrap(); + plan.machines + .first_mut() + .unwrap() + .environment + .insert("PLOYZ_BUILD_ACTIVE_TIMEOUT_SECONDS".into(), "1".into()); + let cluster = Cluster::create(plan).unwrap(); + cluster.wait_ready(Duration::from_secs(60)).await.unwrap(); + let selected = cluster.initialize_first().await.unwrap().id; + cluster.wait_ready(Duration::from_secs(60)).await.unwrap(); + let address = cluster.api_address(0).unwrap(); + let (uploading, mut active) = request(&address, selected).await; + assert!(matches!(event(&mut active).await, Event::Admitted { .. })); + let (disconnected, mut queued) = request(&address, selected).await; + assert!(matches!( + event(&mut queued).await, + Event::Progress(Progress::Stage(Stage::Queued)) + )); + drop(disconnected); + assert!(matches!( + terminal(&mut queued).await, + Outcome::Failed { + stage: Stage::Queued, + .. + } + )); + assert!( + matches!(terminal(&mut active).await, Outcome::Failed { stage: Stage::Upload, message, .. } if message.contains("timeout")) + ); + drop(uploading); + + // Keep an admitted partial upload and another waiting connection across + // restart. Restore the normal active budget for the subsequent real build. + cluster.machine_shell(0, "mv /usr/local/bin/ployzd /usr/local/bin/ployzd-real; printf '%s\n' '#!/bin/sh' 'unset PLOYZ_BUILD_ACTIVE_TIMEOUT_SECONDS' 'exec /usr/local/bin/ployzd-real \"$@\"' > /usr/local/bin/ployzd; chmod +x /usr/local/bin/ployzd").unwrap(); + cluster.restart(0).unwrap(); + cluster.wait_ready(Duration::from_secs(60)).await.unwrap(); + let (interrupted, mut active) = request(&address, selected).await; + assert!(matches!(event(&mut active).await, Event::Admitted { .. })); + interrupted + .send( + remote::encode(&Input::Entry { + path: b"source".to_vec(), + kind: remote::Kind::Directory, + mode: 0o755, + }) + .unwrap(), + ) + .await + .unwrap(); + let (_waiting, mut queued) = request(&address, selected).await; + assert!(matches!( + event(&mut queued).await, + Event::Progress(Progress::Stage(Stage::Queued)) + )); + cluster.restart(0).unwrap(); + for response in [&mut active, &mut queued] { + let result = tokio::time::timeout(Duration::from_secs(10), response.message()) + .await + .unwrap(); + if let Ok(Some(payload)) = result { + assert!(matches!( + remote::decode::(&payload).unwrap(), + Event::Finished(Outcome::Failed { .. } | Outcome::Unknown { .. }) + )); + } + } + cluster.wait_ready(Duration::from_secs(60)).await.unwrap(); + let root = std::env::temp_dir().join(format!("ployz-build-807-{}", uuid::Uuid::new_v4())); + fs::create_dir_all(&root).unwrap(); + fs::write( + root.join("compose.yaml"), + "name: restart\nservices:\n app:\n image: ployz-restart:built\n build: .\n", + ) + .unwrap(); + fs::write( + root.join("Dockerfile"), + "FROM alpine:3.23.3\nCOPY payload /payload\nCMD [\"cat\", \"/payload\"]\n", + ) + .unwrap(); + fs::write(root.join("payload"), "safe-after-restart\n").unwrap(); + let output = tokio::time::timeout( + Duration::from_secs(180), + tokio::process::Command::new(env!("CARGO_BIN_EXE_ployz")) + .current_dir(&root) + .env("PLOYZ_CONFIG", root.join("config.yaml")) + .args([ + "--connect", + &address, + "build", + &format!("--remote={selected}"), + "app", + ]) + .kill_on_drop(true) + .output(), + ) + .await + .unwrap() + .unwrap(); + assert!( + output.status.success(), + "{}", + String::from_utf8_lossy(&output.stderr) + ); + let stdout = String::from_utf8(output.stdout).unwrap(); + let exact = stdout + .split_whitespace() + .find(|word| word.contains("@sha256:")) + .unwrap(); + assert_eq!( + cluster + .machine_shell(0, &format!("docker run --rm {exact}")) + .unwrap(), + "safe-after-restart\n" + ); + fs::remove_dir_all(root).unwrap(); +} diff --git a/crates/ployz/tests/connect/support.rs b/crates/ployz/tests/connect/support.rs index 59be2b65bf..71b280ade7 100644 --- a/crates/ployz/tests/connect/support.rs +++ b/crates/ployz/tests/connect/support.rs @@ -321,8 +321,35 @@ impl MachineRpc for DiscoveryService { .map(|target| target.name.clone()) .collect(), ); + if recorder.queued { + sender + .send(Ok(remote::encode(&Event::Progress( + ployz_build::Progress::Stage(ployz_build::Stage::Queued), + )) + .unwrap())) + .await + .unwrap(); + assert!( + tokio::time::timeout(std::time::Duration::from_millis(50), request.message()) + .await + .is_err(), + "client uploaded before admission" + ); + } + if let Some(outcome) = &recorder.admission_outcome { + let _ = sender + .send(Ok( + remote::encode(&Event::Finished(outcome.clone())).unwrap() + )) + .await; + return; + } sender - .send(Ok(remote::encode(&Event::Admitted { machine_id }).unwrap())) + .send(Ok(remote::encode(&Event::Admitted { + machine_id, + active_timeout: ployz_build::EXECUTION_TIMEOUT, + }) + .unwrap())) .await .unwrap(); let mut upload = remote::Upload::new().unwrap(); @@ -1093,4 +1120,6 @@ pub(super) struct BuildRecorder { pub(super) routes: Mutex>, pub(super) targets: Mutex>>, pub(super) uploads: AtomicUsize, + pub(super) queued: bool, + pub(super) admission_outcome: Option, } diff --git a/crates/ployz/tests/remote_build_cli.rs b/crates/ployz/tests/remote_build_cli.rs index 88bbc79c14..41f9b2eec9 100644 --- a/crates/ployz/tests/remote_build_cli.rs +++ b/crates/ployz/tests/remote_build_cli.rs @@ -1,3 +1,5 @@ +//! Remote Build CLI admission, outcome rendering, and absence of local Docker I/O. + use ployz_core::{BUILD_CAPABILITY, MachineTarget, RoutingRequest}; use std::{ fs, @@ -32,7 +34,10 @@ async fn standalone_remote_build_never_invokes_local_docker_and_keeps_the_servic .insert(BUILD_CAPABILITY.parse().unwrap()); let mut service = support::DiscoveryService::new(description); service.machines = vec![support::machine('a', "tower")]; - let recorder = Arc::new(support::BuildRecorder::default()); + let recorder = Arc::new(support::BuildRecorder { + queued: true, + ..Default::default() + }); service.builds = Some(recorder.clone()); let (address, server) = support::serve_discovery(service).await; let run = |args: &[&str]| { @@ -60,6 +65,13 @@ async fn standalone_remote_build_never_invokes_local_docker_and_keeps_the_servic "{}", String::from_utf8_lossy(&output.stderr) ); + let stderr = String::from_utf8_lossy(&output.stderr); + assert!( + stderr.contains("Queued") + && stderr.contains("queue wait:") + && stderr.contains("execution:"), + "{stderr}" + ); let stdout = String::from_utf8_lossy(&output.stdout); assert!( stdout.contains(support::machine_id('a').as_str()), @@ -154,3 +166,84 @@ async fn standalone_remote_build_never_invokes_local_docker_and_keeps_the_servic server.abort(); fs::remove_dir_all(root).unwrap(); } + +#[tokio::test] +async fn remote_queue_outcomes_name_machine_and_unattempted_work() { + use ployz_build::{Stage, TargetEvidence, WorkEvidence, remote::Outcome}; + let root = std::env::temp_dir().join(format!("ployz-queue-cli-{}", uuid::Uuid::new_v4())); + fs::create_dir_all(&root).unwrap(); + fs::write( + root.join("compose.yaml"), + "name: demo\nservices: {api: {build: .}}\n", + ) + .unwrap(); + fs::write(root.join("Dockerfile"), "FROM scratch\n").unwrap(); + for reason in [ + "queue full", + "queue expired", + "queue cancelled", + "termination unknown", + ] { + let mut description = support::test_description(); + description.machine_id = support::machine_id('a'); + description + .capabilities + .insert(BUILD_CAPABILITY.parse().unwrap()); + let mut service = support::DiscoveryService::new(description); + service.machines = vec![support::machine('a', "tower")]; + let work = WorkEvidence(std::collections::BTreeMap::from([( + "api".into(), + TargetEvidence::Unattempted, + )])); + let outcome = if reason == "termination unknown" { + Outcome::Unknown { + stage: Stage::Admission, + message: reason.into(), + work, + } + } else { + Outcome::Failed { + stage: Stage::Queued, + message: reason.into(), + work, + } + }; + let recorder = Arc::new(support::BuildRecorder { + queued: true, + admission_outcome: Some(outcome), + ..Default::default() + }); + service.builds = Some(recorder.clone()); + let (address, server) = support::serve_discovery(service).await; + let output = 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")) + .args([ + "--connect", + &format!("tcp://{address}"), + "build", + "--remote=tower", + "api", + ]) + .output() + .await + .unwrap(); + assert!(!output.status.success()); + let stderr = String::from_utf8_lossy(&output.stderr); + assert!( + stderr.contains(support::machine_id('a').as_str()) + && stderr.contains(reason) + && stderr.contains("Unattempted"), + "{stderr}" + ); + assert!( + stderr.contains("queue wait:") && stderr.contains("execution: 0.00s"), + "{stderr}" + ); + assert_eq!(recorder.uploads.load(Ordering::SeqCst), 0); + server.abort(); + } + fs::remove_dir_all(root).unwrap(); +} diff --git a/crates/ployzd/src/build.rs b/crates/ployzd/src/build.rs index b9d104be6a..e745ebaef7 100644 --- a/crates/ployzd/src/build.rs +++ b/crates/ployzd/src/build.rs @@ -1,9 +1,12 @@ //! Admission and lifetime of one connection-scoped remote Build. +mod queue; +pub(crate) use queue::Runner; + use crate::logs::RpcStream; use ployz_build::{ Admission, BuildError, HostPolicy, Output, Progress, Stage, - remote::{self, Definition, Event, Input, Outcome, Upload}, + remote::{self, Definition, Event, Input, Outcome}, }; use ployz_core::{MachineId, OpaquePayload}; use std::time::Duration; @@ -14,11 +17,11 @@ use tonic::Status; pub(crate) fn start( machine_id: MachineId, requests: impl Stream> + Send + Unpin + 'static, - policy: HostPolicy, + runner: std::sync::Arc, ) -> RpcStream { let (events, receiver) = mpsc::channel(8); tokio::spawn(async move { - let outcome = attempt(machine_id, requests, &events, policy).await; + let outcome = attempt(machine_id, requests, &events, runner).await; let message = "Build terminal report exceeds the response size limit"; let fallback = match &outcome { Outcome::Unknown { stage, .. } => Outcome::Unknown { @@ -43,7 +46,7 @@ async fn attempt( machine_id: MachineId, mut requests: impl Stream> + Send + Unpin + 'static, events: &mpsc::Sender>, - policy: HostPolicy, + runner: std::sync::Arc, ) -> Outcome { let start = tokio::time::timeout(Duration::from_secs(10), requests.next()).await; let definition = match start { @@ -59,6 +62,36 @@ async fn attempt( "Build must name between one and 128 targets", ); } + let work = ployz_build::WorkEvidence::new(&definition.targets); + let permit = match runner.enter() { + Ok(queue::Entry::Active(permit)) => permit, + Ok(queue::Entry::Waiting(waiting)) => { + let queued = remote::encode(&Event::Progress(Progress::Stage(Stage::Queued))) + .expect("bounded queue frame"); + if events.send(Ok(queued)).await.is_err() { + return failed(Stage::Queued, "Build client disconnected").with_work(work); + } + tokio::select! { + biased; + () = runner.shutdown.cancelled() => return failed(Stage::Queued, "Build daemon stopped; execution was not attempted").with_work(work), + () = events.closed() => return failed(Stage::Queued, "Build client disconnected").with_work(work), + frame = requests.next() => { + let reason = match frame { + Some(Ok(payload)) if matches!(remote::decode::(&payload), Ok(Input::Cancel)) => "Build cancelled while queued; execution was not attempted", + None | Some(Err(_)) => "Build client disconnected while queued; execution was not attempted", + Some(Ok(_)) => "Build upload before admission is forbidden; execution was not attempted", + }; + return failed(Stage::Queued, reason).with_work(work); + } + admitted = runner.admit(waiting) => match admitted { + Ok(permit) => permit, + Err(reason) => return failed(Stage::Queued, reason.to_string()).with_work(work), + }, + } + } + Err(reason) => return failed(Stage::Queued, reason.to_string()).with_work(work), + }; + let policy = runner.policy.clone(); let admission = match tokio::task::spawn_blocking({ let policy = policy.clone(); move || Admission::try_acquire_with(&policy) @@ -66,7 +99,7 @@ async fn attempt( .await { Ok(Ok(admission)) => admission, - Ok(Err(error)) => return failure(Stage::Admission, error), + 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( @@ -76,14 +109,18 @@ async fn attempt( let cancellation = admission.cancellation(); let remaining = admission.remaining(); let deadline = tokio::time::Instant::now() + remaining; - let admitted = - remote::encode(&Event::Admitted { machine_id }).expect("bounded admission frame"); + let admitted = remote::encode(&Event::Admitted { + machine_id, + active_timeout: policy.active_timeout, + }) + .expect("bounded admission frame"); if events.send(Ok(admitted)).await.is_err() { return failed(Stage::Admission, "Build client disconnected"); } let (upload, source) = mpsc::channel(2); let output = events.clone(); let mut execution = tokio::task::spawn_blocking(move || { + let _permit = permit; receive_and_execute( machine_id, definition, source, admission, policy, &output, &observed, ) @@ -96,6 +133,7 @@ async fn attempt( tokio::select! { result = &mut execution => return joined(result).with_work(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(), () = tokio::time::sleep_until(deadline) => "Build active timeout expired".to_owned(), } @@ -105,10 +143,13 @@ async fn attempt( // confirm its end keeps both its admission and persistent quarantine. let outcome = match tokio::time::timeout(Duration::from_secs(70), execution).await { Ok(result) => match joined(result) { - outcome @ (Outcome::Images { .. } - | Outcome::Validated { .. } - | Outcome::Published { .. } - | Outcome::Unknown { .. }) => outcome, + outcome @ Outcome::Unknown { .. } => outcome, + Outcome::Images { .. } | Outcome::Validated { .. } | Outcome::Published { .. } => { + failed( + Stage::Cleanup, + format!("{reason}; output completed before cancellation was observed"), + ) + } Outcome::Failed { stage, message, @@ -159,14 +200,12 @@ fn receive_and_execute( evidence: &std::sync::Mutex, ) -> Outcome { let cancellation = admission.cancellation(); - let mut upload = match Upload::new() { + let deadline = std::time::Instant::now() + admission.remaining(); + let mut upload = match admission.upload() { Ok(upload) => upload, Err(error) => return failed(Stage::Upload, error.to_string()), }; loop { - if let Err(error) = admission.check() { - return failure(Stage::Upload, error); - } let Some(payload) = source.blocking_recv() else { return failed( Stage::Upload, @@ -185,11 +224,6 @@ fn receive_and_execute( break; } } - let upload = match upload.complete() { - Ok(upload) => upload, - Err(error) => return failed(Stage::Upload, error.to_string()), - }; - let deadline = std::time::Instant::now() + admission.remaining(); let progress = |progress| { evidence .lock() @@ -214,7 +248,7 @@ fn receive_and_execute( } }; progress(Progress::Stage(Stage::Upload)); - match upload.execute(&definition, admission, Some(&policy.docker), &progress) { + 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 }, diff --git a/crates/ployzd/src/build/queue.rs b/crates/ployzd/src/build/queue.rs new file mode 100644 index 0000000000..5d1fb1c024 --- /dev/null +++ b/crates/ployzd/src/build/queue.rs @@ -0,0 +1,167 @@ +//! Machine-local FIFO admission. Permits follow execution, not the RPC task. + +use ployz_build::{BuildError, HostPolicy}; +use std::{ + future::Future, + pin::Pin, + sync::Arc, + task::{Context, Poll, Waker}, +}; +use tokio::{ + sync::{AcquireError, OwnedSemaphorePermit, Semaphore}, + time::Instant, +}; +use tokio_util::sync::CancellationToken; + +pub(crate) struct Runner { + pub(super) policy: HostPolicy, + active: Arc, + waiting: Arc, + pub(super) shutdown: CancellationToken, +} + +impl Runner { + pub(crate) fn new( + policy: HostPolicy, + shutdown: CancellationToken, + ) -> Result, BuildError> { + policy.validate()?; + Ok(Arc::new(Self { + waiting: Arc::new(Semaphore::new(policy.queue_capacity)), + active: Arc::new(Semaphore::new(1)), + policy, + shutdown, + })) + } + + pub(super) fn enter( + &self, + ) -> Result< + Entry> + Send>, + QueueError, + > { + if self.shutdown.is_cancelled() { + return Err(QueueError::Stopping); + } + let mut admission = Box::pin(self.active.clone().acquire_owned()); + // Enroll synchronously so a release or a slow progress consumer cannot + // let a later arrival overtake this waiter before its first await. + match admission + .as_mut() + .poll(&mut Context::from_waker(Waker::noop())) + { + Poll::Ready(result) => result.map(Entry::Active).map_err(|_| QueueError::Stopping), + Poll::Pending => { + let capacity = self + .waiting + .clone() + .try_acquire_owned() + .map_err(|_| QueueError::Full)?; + Ok(Entry::Waiting(Waiting { + _capacity: capacity, + admission, + expires: Instant::now() + self.policy.queue_timeout, + })) + } + } + } + + pub(super) async fn admit( + &self, + waiting: Waiting, + ) -> Result + where + F: Future>, + { + tokio::select! { + biased; + () = tokio::time::sleep_until(waiting.expires) => Err(QueueError::Expired), + result = waiting.admission => result.map_err(|_| QueueError::Stopping), + } + } +} + +#[derive(Debug, thiserror::Error)] +pub(super) enum QueueError { + #[error("Build queue is full; execution was not attempted")] + Full, + #[error("Build queue timeout expired; execution was not attempted")] + Expired, + #[error("Build daemon is stopping; execution was not attempted")] + Stopping, +} + +pub(super) enum Entry { + Active(OwnedSemaphorePermit), + Waiting(Waiting), +} + +pub(super) struct Waiting { + _capacity: OwnedSemaphorePermit, + admission: Pin>, + expires: Instant, +} + +#[cfg(test)] +mod tests { + use super::*; + use futures_util::poll; + use std::{task::Poll, time::Duration}; + + fn active(runner: &Runner) -> OwnedSemaphorePermit { + let Entry::Active(permit) = runner.enter().unwrap() else { + panic!("expected admission") + }; + permit + } + fn waiting(runner: &Runner) -> impl Future> { + let Entry::Waiting(waiting) = runner.enter().unwrap() else { + panic!("expected waiting") + }; + runner.admit(waiting) + } + + #[tokio::test(start_paused = true)] + async fn enrolled_waiter_cannot_be_overtaken_before_it_is_polled() { + let runner = Runner::new(HostPolicy::default(), CancellationToken::new()).unwrap(); + let owner = active(&runner); + let _first = runner.enter().unwrap(); + drop(owner); + assert!( + matches!(runner.enter(), Ok(Entry::Waiting(_))), + "new arrival overtook an enrolled waiter" + ); + } + + #[tokio::test(start_paused = true)] + async fn fifo_bounds_expiry_and_cancelled_waiters_release_capacity() { + let runner = Runner::new(HostPolicy::default(), CancellationToken::new()).unwrap(); + let owner = active(&runner); + let mut entries = (0..8).map(|_| waiting(&runner)).collect::>(); + assert!(matches!(runner.enter(), Err(QueueError::Full))); + let mut first = Box::pin(entries.remove(0)); + let mut second = Box::pin(entries.remove(0)); + assert!(poll!(&mut first).is_pending()); + assert!(poll!(&mut second).is_pending()); + drop(owner); + assert!(poll!(&mut second).is_pending()); + let Poll::Ready(Ok(owner)) = poll!(&mut first) else { + panic!("FIFO head did not get admission") + }; + // Cancelling a queued future removes both its FIFO entry and capacity. + drop(second); + drop(entries); + let mut expiring = Box::pin(waiting(&runner)); + assert!(poll!(&mut expiring).is_pending()); + tokio::time::advance(Duration::from_secs(599)).await; + assert!(poll!(&mut expiring).is_pending()); + tokio::time::advance(Duration::from_secs(1)).await; + assert!(matches!( + poll!(&mut expiring), + Poll::Ready(Err(QueueError::Expired)) + )); + assert!(matches!(runner.enter(), Ok(Entry::Waiting(_)))); + drop(owner); + drop(active(&runner)); + } +} diff --git a/crates/ployzd/src/build/tests.rs b/crates/ployzd/src/build/tests.rs index 27fa629338..821abfdaa7 100644 --- a/crates/ployzd/src/build/tests.rs +++ b/crates/ployzd/src/build/tests.rs @@ -27,6 +27,9 @@ struct Fixture { impl Fixture { async fn new() -> Self { + Self::with_policy(|_| {}).await + } + async fn with_policy(update: impl FnOnce(&mut HostPolicy)) -> Self { let root = std::env::temp_dir().join(format!("ployz-remote-build-test-{}", MachineId::random())); fs::create_dir_all(&root).unwrap(); @@ -60,13 +63,15 @@ impl Fixture { Some((replicated, crate::corrosion::AdminClient::new("/no/admin"))), ) .with_optional_containers(Some(runtime)); - let policy = HostPolicy { + let mut policy = HostPolicy { state_directory: root.clone(), docker: root.join("docker"), active_timeout: Duration::from_secs(30), configuration_file: root.join("build.yaml"), + ..Default::default() }; - service.build_policy = policy.clone(); + update(&mut policy); + service.builds = Runner::new(policy.clone(), Default::default()).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()); @@ -155,15 +160,13 @@ async fn terminal(response: &mut tonic::Streaming) -> Outcome { } #[tokio::test] -async fn admitted_upload_refuses_competitors_and_disconnect_releases_unused_ownership() { +async fn admitted_upload_queues_competitors_and_disconnect_releases_unused_ownership() { let fixture = Fixture::new().await; let (first, mut response) = fixture.request(Output::Load).await; assert!(matches!(event(&mut response).await, Event::Admitted { .. })); assert!(!fixture.root.join("executed").exists()); - let (_second, mut rejected) = fixture.request(Output::Load).await; - assert!( - matches!(terminal(&mut rejected).await, Outcome::Failed { stage: Stage::Admission, message, .. } if message.contains("busy")) - ); + let (second, mut queued) = fixture.request(Output::Load).await; + assert!(matches!(event(&mut queued).await, Event::Progress(_))); first .send( remote::encode(&Input::Entry { @@ -184,6 +187,9 @@ async fn admitted_upload_refuses_competitors_and_disconnect_releases_unused_owne } )); assert!(!fixture.root.join("executed").exists()); + assert!(matches!(event(&mut queued).await, Event::Admitted { .. })); + drop(second); + let _ = terminal(&mut queued).await; let (_third, mut admitted) = fixture.request(Output::Load).await; assert!(matches!(event(&mut admitted).await, Event::Admitted { .. })); } @@ -386,7 +392,11 @@ async fn upload_timeout_stops_before_execution_and_releases_admission() { .unwrap())) .await .unwrap(); - let mut responses = start(fixture.machine.id, ReceiverStream::new(receiver), policy); + let mut responses = start( + fixture.machine.id, + ReceiverStream::new(receiver), + Runner::new(policy, Default::default()).unwrap(), + ); assert!(matches!( remote::decode::(&responses.next().await.unwrap().unwrap()).unwrap(), Event::Admitted { .. } @@ -519,6 +529,131 @@ async fn registry_publication_keeps_completed_and_unattempted_targets() { } } +#[tokio::test] +async fn queued_disconnect_cancellation_expiry_and_full_queue_never_upload() { + let fixture = Fixture::with_policy(|policy| { + policy.queue_capacity = 1; + policy.queue_timeout = Duration::from_millis(200); + }) + .await; + let (active, mut response) = fixture.request(Output::Load).await; + assert!(matches!(event(&mut response).await, Event::Admitted { .. })); + for action in ["disconnect", "cancel", "early-upload", "expire"] { + let (waiting, mut queued) = fixture.request(Output::Load).await; + assert!(matches!( + event(&mut queued).await, + Event::Progress(Progress::Stage(Stage::Queued)) + )); + let (_full, mut rejected) = fixture.request(Output::Load).await; + assert!( + matches!(terminal(&mut rejected).await, Outcome::Failed { stage: Stage::Queued, message, work } if message.contains("full") && work.0.get("api") == Some(&ployz_build::TargetEvidence::Unattempted)) + ); + match action { + "disconnect" => drop(waiting), + "cancel" => waiting + .send(remote::encode(&Input::Cancel).unwrap()) + .await + .unwrap(), + "early-upload" => waiting + .send(remote::encode(&Input::Data(vec![1])).unwrap()) + .await + .unwrap(), + "expire" => {} + _ => unreachable!(), + } + let Outcome::Failed { + stage: Stage::Queued, + message, + work, + } = terminal(&mut queued).await + else { + panic!("queued work ran") + }; + let expected = match action { + "disconnect" => "disconnected", + "cancel" => "cancelled", + "early-upload" => "before admission", + "expire" => "expired", + _ => unreachable!(), + }; + assert!(message.contains(expected), "{message}"); + assert_eq!( + work.0.get("api"), + Some(&ployz_build::TargetEvidence::Unattempted) + ); + assert!(!fixture.root.join("executed").exists()); + } + drop(active); + let _ = terminal(&mut response).await; +} + +#[tokio::test] +async fn captured_client_cancels_in_queue_without_uploading() { + let fixture = Fixture::new().await; + let (_active, mut response) = fixture.request(Output::Load).await; + assert!(matches!(event(&mut response).await, Event::Admitted { .. })); + let capture = fixture.capture(); + let client = ployz::connect::connect( + Path::new("/missing-test-config"), + Some(&fixture.address.replace("http://", "tcp://")), + None, + ) + .await + .unwrap(); + let cancellation = tokio_util::sync::CancellationToken::new(); + let result = capture + .execute_remote(&client, fixture.machine.id, cancellation.clone(), |event| { + if matches!(event, Progress::Stage(Stage::Queued)) { + cancellation.cancel(); + } + assert!(!matches!(event, Progress::Stage(Stage::Upload))); + }) + .await; + assert!( + matches!(result, Outcome::Failed { stage: Stage::Queued, message, .. } if message.contains("cancelled")) + ); + assert!(!fixture.root.join("executed").exists()); +} + +#[tokio::test] +async fn client_waits_past_connection_deadline_and_uploads_only_after_admission() { + let fixture = Fixture::new().await; + let (active, mut response) = fixture.request(Output::Load).await; + assert!(matches!(event(&mut response).await, Event::Admitted { .. })); + let capture = fixture.capture(); + let client = ployz::connect::connect( + Path::new("/missing-test-config"), + Some(&fixture.address.replace("http://", "tcp://")), + None, + ) + .await + .unwrap(); + let (waiting, mut observed) = mpsc::channel(1); + let selected = fixture.machine.id; + let execution = tokio::spawn(async move { + capture + .execute_remote(&client, selected, Default::default(), |event| { + if matches!(event, Progress::Stage(Stage::Queued)) { + waiting.try_send(()).unwrap(); + } + }) + .await + }); + observed.recv().await.unwrap(); + tokio::time::sleep(Duration::from_secs(11)).await; + assert!( + !execution.is_finished(), + "client abandoned the configured queue wait" + ); + assert!( + !fixture.root.join("build-upload/source").exists(), + "queued capture uploaded early" + ); + drop(active); + let _ = terminal(&mut response).await; + assert!(matches!(execution.await.unwrap(), Outcome::Images { .. })); +} + #[tokio::test] async fn host_configuration_and_cache_clearing_share_remote_admission() { let fixture = Fixture::new().await; diff --git a/crates/ployzd/src/daemon.rs b/crates/ployzd/src/daemon.rs index 0ee56de050..bc905194f5 100644 --- a/crates/ployzd/src/daemon.rs +++ b/crates/ployzd/src/daemon.rs @@ -99,8 +99,26 @@ impl Daemon { /// /// If construction, binding, or required planes fail. pub async fn start(config: DaemonConfig) -> Result { + let build_policy = ployz_build::HostPolicy::from_environment() + .map_err(|error| io::Error::new(io::ErrorKind::InvalidInput, error))?; + Self::start_with_build_policy(config, build_policy).await + } + + async fn start_with_build_policy( + config: DaemonConfig, + build_policy: ployz_build::HostPolicy, + ) -> Result { let store = Arc::new(Mutex::new(LocalMachineStore::open(&config.data_dir)?)); let socket_lock = claim_socket(&config.socket)?; + let cleanup = tokio::task::spawn_blocking({ + let policy = build_policy.clone(); + move || ployz_build::Admission::cleanup_abandoned(&policy) + }) + .await + .map_err(io::Error::other)?; + if let Err(error) = cleanup { + eprintln!("WARNING: abandoned Build cleanup: {error}"); + } let local_record = store .lock() .map_err(|_| Error::StorePoisoned)? @@ -167,6 +185,10 @@ impl Daemon { .unwrap_or_else(|| Path::new("/run/ployz")) .join("ingress"); let machine_api = MachineApi::builder(Arc::clone(&store), reset.clone()) + .with_builds( + crate::build::Runner::new(build_policy, shutdown.clone()) + .map_err(|error| io::Error::new(io::ErrorKind::InvalidInput, error))?, + ) .with_participation(participating.clone()) .with_cluster( corrosion @@ -765,6 +787,48 @@ mod tests { response.decode::().unwrap(); } + #[tokio::test] + async fn startup_cleans_abandoned_builder_without_a_build_request() { + use std::os::unix::fs::PermissionsExt as _; + let root = TestDir::new("ployzd-builder-restart"); + let (config, socket) = test_config(&root.0, ContainerMode::Absent); + let policy = ployz_build::HostPolicy { + state_directory: root.0.clone(), + docker: root.0.join("docker"), + ..Default::default() + }; + let marker = root.0.join(format!("{}.lock", ployz_build::builder_name())); + fs::write(&marker, "termination unconfirmed\n").unwrap(); + fs::write( + &policy.docker, + "#!/bin/sh\nprintf '%s\\n' \"$*\" > cleaned\n", + ) + .unwrap(); + fs::set_permissions(&policy.docker, fs::Permissions::from_mode(0o700)).unwrap(); + let daemon = Daemon::start_with_build_policy(config, policy.clone()) + .await + .unwrap(); + assert!( + fs::read_to_string(root.0.join("cleaned")) + .unwrap() + .contains("buildx rm") + ); + assert!(!fs::read_to_string(marker).unwrap().is_empty()); + assert!( + ployz_build::Admission::try_acquire_with(&policy) + .err() + .unwrap() + .is_unknown() + ); + assert!( + describe(&socket) + .await + .supports(DESCRIBE_CONTRACT_CAPABILITY) + ); + daemon.request_stop(); + daemon.wait().await.unwrap(); + } + #[tokio::test] async fn absent_containers_start_without_container_capabilities() { let root = TestDir::new("ployzd-daemon-absent"); diff --git a/crates/ployzd/src/machine_api/local.rs b/crates/ployzd/src/machine_api/local.rs index e129d94583..d10b1db758 100644 --- a/crates/ployzd/src/machine_api/local.rs +++ b/crates/ployzd/src/machine_api/local.rs @@ -44,7 +44,7 @@ pub struct MachineService { allocator_endpoint: Option<(MachineId, std::net::SocketAddr)>, cloud_pairing: Option>>, runtime_watch: Arc, - pub(crate) build_policy: ployz_build::HostPolicy, + pub(crate) builds: Arc, } impl MachineService { @@ -64,7 +64,8 @@ impl MachineService { allocator_endpoint: None, cloud_pairing: None, runtime_watch: Arc::default(), - build_policy: ployz_build::HostPolicy::default(), + builds: crate::build::Runner::new(Default::default(), Default::default()) + .expect("default Build policy"), } } @@ -597,7 +598,7 @@ impl MachineRpc for MachineService { Ok(Response::new(crate::build::start( self.local_record()?.id(), request.into_inner(), - self.build_policy.clone(), + self.builds.clone(), ))) } diff --git a/crates/ployzd/src/machine_api/mod.rs b/crates/ployzd/src/machine_api/mod.rs index cb7d6a98aa..6bbe6cc4e1 100644 --- a/crates/ployzd/src/machine_api/mod.rs +++ b/crates/ployzd/src/machine_api/mod.rs @@ -68,6 +68,11 @@ impl MachineApi { } impl MachineApiBuilder { + pub(crate) fn with_builds(mut self, builds: Arc) -> Self { + self.service.builds = builds; + self + } + #[must_use] pub(crate) fn with_participation(mut self, participating: watch::Sender) -> Self { self.service = self.service.with_participation(participating);