From 38a397fa20d1a3239dd79cab6cad51ef2600cfac Mon Sep 17 00:00:00 2001 From: Maofeng Date: Thu, 30 Jul 2026 07:18:07 +0800 Subject: [PATCH] fix(package): make dev deployments retryable Track the source fingerprint captured before compilation so edits made during a build trigger another cycle. Keep successful build artifacts pending while the Console is unavailable and retry deployment without recompiling. Make activation retries idempotent by pruning every stale matching development revision after the active revision is switched. --- crates/wasmeld-package/src/dev.rs | 200 ++++++++++++++++-------------- 1 file changed, 107 insertions(+), 93 deletions(-) diff --git a/crates/wasmeld-package/src/dev.rs b/crates/wasmeld-package/src/dev.rs index 048ee29..98cb5d0 100644 --- a/crates/wasmeld-package/src/dev.rs +++ b/crates/wasmeld-package/src/dev.rs @@ -20,6 +20,7 @@ use crate::{BuildProfile, PackRequest, PackageIdentity, PackedComponent, pack_co const DEFAULT_CONSOLE: &str = "http://127.0.0.1:8080"; const DEFAULT_POLL_MS: u64 = 350; +const DEPLOY_RETRY_INTERVAL: Duration = Duration::from_secs(1); struct DevOptions { manifest_path: PathBuf, @@ -42,25 +43,21 @@ struct ServiceIdentity { revision: String, } -#[derive(Debug, Deserialize)] -struct Deployment { - active_revision: String, -} - pub(crate) fn run(arguments: Vec) -> Result<(), Box> { let options = parse_options(arguments)?; let manifest_path = fs::canonicalize(&options.manifest_path)?; let watch_roots = watch_roots(&manifest_path)?; let client = Client::builder() .user_agent(format!("wasmeld-dev/{}", env!("CARGO_PKG_VERSION"))) + .connect_timeout(Duration::from_secs(2)) .timeout(Duration::from_secs(120)) .build()?; - ensure_console(&client, &options.console)?; let temporary_dir = env::temp_dir().join(format!("wasmeld-dev-{}", std::process::id())); fs::create_dir_all(&temporary_dir)?; let output = temporary_dir.join("component.wasmpkg"); let mut observed = None; + let mut pending_deployment = None; println!("watching:"); for root in &watch_roots { @@ -70,33 +67,96 @@ pub(crate) fn run(arguments: Vec) -> Result<(), Box { - println!( - "ready: {}@{} ({})", - packed.manifest.id, packed.manifest.revision, packed.manifest.sha256 - ); - if options.once { - return Ok(()); + if observed.as_ref() != Some(¤t) { + // Record the fingerprint that started this build. A save that + // arrives during compilation will differ on the next loop and + // immediately queue another build. + observed = Some(current); + pending_deployment = None; + println!("change detected; building Component"); + match build_component(&options, &manifest_path, &output) { + Ok(packed) => pending_deployment = Some(packed), + Err(error) if options.once => return Err(error), + Err(error) => { + eprintln!("wasmeld dev: {error}"); + eprintln!( + "waiting for the next source change; the previous deployment is intact" + ); } } - Err(error) if options.once => return Err(error), - Err(error) => { - eprintln!("wasmeld dev: {error}"); - eprintln!("waiting for the next source change; the previous deployment is intact"); + } + + if let Some(packed) = pending_deployment.as_ref() { + match deploy_component(&client, &options.console, packed) { + Ok(()) => { + let packed = pending_deployment.take().expect("pending deployment"); + println!( + "ready: {}@{} ({})", + packed.manifest.id, packed.manifest.revision, packed.manifest.sha256 + ); + if options.once { + return Ok(()); + } + } + Err(error) if options.once => return Err(error), + Err(error) => { + eprintln!("wasmeld dev: {error}"); + eprintln!("deployment will retry; the previous deployment is intact"); + thread::sleep(DEPLOY_RETRY_INTERVAL); + continue; + } } } + + thread::sleep(options.poll_interval); } } +fn build_component( + options: &DevOptions, + manifest_path: &Path, + output: &Path, +) -> Result> { + pack_component(PackRequest { + manifest_path: manifest_path.to_owned(), + output: Some(output.to_owned()), + no_build: false, + locked: options.locked, + profile: if options.release { + BuildProfile::Release + } else { + BuildProfile::Debug + }, + identity: PackageIdentity::Development { + id: options.service_id.clone(), + }, + }) +} + +fn deploy_component( + client: &Client, + console: &str, + packed: &PackedComponent, +) -> Result<(), Box> { + register(client, console, packed)?; + activate( + client, + console, + &packed.manifest.id, + &packed.manifest.revision, + )?; + + for revision in stale_development_revisions(client, console, packed)? { + if let Err(error) = unregister(client, console, &packed.manifest.id, &revision) { + eprintln!( + "wasmeld dev: deployed successfully but could not unregister {}@{}: {error}", + packed.manifest.id, revision + ); + } + } + Ok(()) +} + fn parse_options(arguments: Vec) -> Result> { let mut arguments = arguments.into_iter(); let manifest_path = arguments.next().ok_or("dev requires a Cargo.toml path")?; @@ -142,70 +202,21 @@ fn parse_options(arguments: Vec) -> Result Result> { - let packed = pack_component(PackRequest { - manifest_path: manifest_path.to_owned(), - output: Some(output.to_owned()), - no_build: false, - locked: options.locked, - profile: if options.release { - BuildProfile::Release - } else { - BuildProfile::Debug - }, - identity: PackageIdentity::Development { - id: options.service_id.clone(), - }, - })?; - let previous = active_revision(client, &options.console, &packed.manifest.id)?; - register(client, &options.console, &packed)?; - activate( - client, - &options.console, - &packed.manifest.id, - &packed.manifest.revision, - )?; - - if let Some(previous) = previous - && previous != packed.manifest.revision - && is_matching_dev_revision(&previous, &packed.manifest.revision) - && let Err(error) = unregister(client, &options.console, &packed.manifest.id, &previous) - { - eprintln!( - "wasmeld dev: deployed successfully but could not unregister {}@{}: {error}", - packed.manifest.id, previous - ); - } - Ok(packed) -} - -fn ensure_console(client: &Client, console: &str) -> Result<(), Box> { - let response = client.get(format!("{console}/healthz")).send()?; - if response.status().is_success() { - Ok(()) - } else { - Err(http_error("Console health check", response)) - } -} - -fn active_revision( +fn stale_development_revisions( client: &Client, console: &str, - service_id: &str, -) -> Result, Box> { - let response = client - .get(format!("{console}/api/v1/deployments/{service_id}")) - .send()?; - match response.status() { - status if status.is_success() => Ok(Some(response.json::()?.active_revision)), - StatusCode::NOT_FOUND => Ok(None), - _ => Err(http_error("deployment lookup", response)), - } + packed: &PackedComponent, +) -> Result, Box> { + Ok(services(client, console)? + .services + .into_iter() + .filter(|service| { + service.id == packed.manifest.id + && service.revision != packed.manifest.revision + && is_matching_dev_revision(&service.revision, &packed.manifest.revision) + }) + .map(|service| service.revision) + .collect()) } fn register( @@ -243,15 +254,18 @@ fn service_exists( service_id: &str, revision: &str, ) -> Result> { + Ok(services(client, console)? + .services + .iter() + .any(|service| service.id == service_id && service.revision == revision)) +} + +fn services(client: &Client, console: &str) -> Result> { let response = client.get(format!("{console}/api/v1/services")).send()?; if !response.status().is_success() { return Err(http_error("service lookup", response)); } - Ok(response - .json::()? - .services - .iter() - .any(|service| service.id == service_id && service.revision == revision)) + Ok(response.json()?) } fn activate(