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.
2026-07-30 07:18:07 +08:00
parent 2650b8ef14
commit 38a397fa20
1 changed files with 107 additions and 93 deletions
@@ -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<String>) -> Result<(), Box<dyn std::error::Error>> {
    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<String>) -> Result<(), Box<dyn std::error::Erro

    loop {
        let current = source_fingerprint(&watch_roots)?;
-        if observed.as_ref() == Some(&current) {
-            thread::sleep(options.poll_interval);
-            continue;
-        }
-
-        println!("change detected; building Component");
-        let result = build_and_deploy(&client, &options, &manifest_path, &output);
-        observed = Some(source_fingerprint(&watch_roots)?);
-        match result {
-            Ok(packed) => {
-                println!(
-                    "ready: {}@{} ({})",
-                    packed.manifest.id, packed.manifest.revision, packed.manifest.sha256
-                );
-                if options.once {
-                    return Ok(());
+        if observed.as_ref() != Some(&current) {
+            // 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<PackedComponent, Box<dyn std::error::Error>> {
+    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<dyn std::error::Error>> {
+    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<String>) -> Result<DevOptions, Box<dyn std::error::Error>> {
    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<String>) -> Result<DevOptions, Box<dyn std::erro
    })
 }

-fn build_and_deploy(
-    client: &Client,
-    options: &DevOptions,
-    manifest_path: &Path,
-    output: &Path,
-) -> Result<PackedComponent, Box<dyn std::error::Error>> {
-    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<dyn std::error::Error>> {
-    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<Option<String>, Box<dyn std::error::Error>> {
-    let response = client
-        .get(format!("{console}/api/v1/deployments/{service_id}"))
-        .send()?;
-    match response.status() {
-        status if status.is_success() => Ok(Some(response.json::<Deployment>()?.active_revision)),
-        StatusCode::NOT_FOUND => Ok(None),
-        _ => Err(http_error("deployment lookup", response)),
-    }
+    packed: &PackedComponent,
+) -> Result<Vec<String>, Box<dyn std::error::Error>> {
+    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<bool, Box<dyn std::error::Error>> {
+    Ok(services(client, console)?
+        .services
+        .iter()
+        .any(|service| service.id == service_id && service.revision == revision))
+}
+
+fn services(client: &Client, console: &str) -> Result<ServiceList, Box<dyn std::error::Error>> {
    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::<ServiceList>()?
-        .services
-        .iter()
-        .any(|service| service.id == service_id && service.revision == revision))
+    Ok(response.json()?)
 }

 fn activate(