fix: locate coursier artifacts by coordinate so private maven registries work (#10501)

* fix: locate coursier artifacts by coordinate so private maven registries work Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * fix: match maven coordinates by path component, longest first Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * fix: skip the empty directory a 404 leaves at a maven coordinate Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * fix: ignore coursier's dot-prefixed bookkeeping when claiming a coordinate Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-09-12 08:05:44 +00:00 · 2026-08-04 11:38:33 +00:00
parent 6f0aa70ae1
commit 56d53d3f32
2 changed files with 307 additions and 37 deletions
@@ -1,4 +1,8 @@
-use std::{collections::HashMap, path::PathBuf, process::Stdio};
+use std::{
+    collections::HashMap,
+    path::{Path, PathBuf},
+    process::Stdio,
+};

 use crate::global_cache::save_cache;
 use anyhow::{anyhow, bail};
@@ -23,12 +27,13 @@ use windmill_queue::{append_logs, CanceledBy, MiniPulledJob};
 use crate::{
    common::{
        build_command_with_isolation, create_args_and_out_file, get_reserved_variables,
-        read_result, resolve_nsjail_timeout, resolve_nsjail_tmp_mount_block, start_child_process,
-        OccupancyMetrics,
+        read_result, resolve_job_timeout, resolve_nsjail_timeout, resolve_nsjail_tmp_mount_block,
+        start_child_process, OccupancyMetrics,
    },
    handle_child, is_sandboxing_enabled, read_ee_registry_bool_with_workspace_override,
    read_ee_registry_with_workspace_override,
    universal_pkg_installer::{par_install_language_dependencies_all_at_once, RequiredDependency},
+    worker_utils::JobPingHeartbeat,
    COURSIER_CACHE_DIR, DISABLE_NUSER, JAVA_CACHE_DIR, JAVA_HOME_DIR, JAVA_REPOSITORY_DIR,
    MAVEN_REPOS, NO_DEFAULT_MAVEN, NSJAIL_PATH, PATH_ENV, PROXY_ENVS,
 };
@@ -294,7 +299,21 @@ pub async fn resolve<'a>(
                    std::env::var("TMP").unwrap_or_else(|_| String::from("/tmp")),
                );
        }
-        let output = cmd.output().await?;
+        // The lockfile has to be read from a clean stdout, so this cannot go through handle_child's
+        // polling loop. That leaves resolution with neither a ping nor a time bound of its own: it
+        // needs the heartbeat not to be reaped as a zombie mid-resolution, and the timeout so a
+        // wedged registry connection cannot park the job in `running` forever.
+        cmd.kill_on_drop(true);
+        let (timeout, ..) = resolve_job_timeout(conn, w_id, *job_id, None).await;
+        let _heartbeat = JobPingHeartbeat::start(conn, *job_id, "java lockfile resolution");
+        let output = tokio::time::timeout(timeout, cmd.output())
+            .await
+            .map_err(|_| {
+                Error::ExecutionErr(format!(
+                    "resolving the java lockfile timed out after {}s",
+                    timeout.as_secs()
+                ))
+            })??;
        // Check if the command was successful
        if output.status.success() {
            String::from_utf8(output.stdout).expect("Failed to convert output to String")
@@ -371,8 +390,8 @@ async fn install<'a>(
    );
    let job_dir = job_dir.to_owned();
    let fetch_dir = format!("{}/tmp-fetch-{}", *JAVA_CACHE_DIR, Uuid::new_v4());
-    let fetch_dir2 = fetch_dir.clone();
-    par_install_language_dependencies_all_at_once(
+    let (cmd_fetch_dir, postinstall_fetch_dir) = (fetch_dir.clone(), fetch_dir.clone());
+    let installed = par_install_language_dependencies_all_at_once(
        deps,
        "java",
        "java",
@@ -431,7 +450,7 @@ async fn install<'a>(
                "--parallel",
                &format!("{}", *JAVA_CONCURRENT_DOWNLOADS),
                "--cache",
-                &fetch_dir,
+                &cmd_fetch_dir,
            ])
            .args(&repos)
            .arg("--intransitive")
@@ -450,35 +469,11 @@ async fn install<'a>(

            Ok(cmd)
        },
-        async move |_| {
-            move_to_repository(&fetch_dir2, 0).await?;
-            remove_dir_all(&fetch_dir2).await?;
-            #[async_recursion]
-            async fn move_to_repository(path: &str, depth: u8) -> anyhow::Result<()> {
-                if depth == 3 {
-                    copy_dir_recursively(
-                        &PathBuf::from(path),
-                        &PathBuf::from(&*JAVA_REPOSITORY_DIR),
-                    )?;
-
-                    return Ok(());
-                }
-                let mut entries = tokio::fs::read_dir(path).await?;
-                loop {
-                    let Some(entry) = entries.next_entry().await? else {
-                        break Ok(());
-                    };
-
-                    let path = entry
-                        .path()
-                        .to_str()
-                        .ok_or(anyhow!("Internal Error: Cannot convert Path to Str"))?
-                        .to_owned();
-
-                    move_to_repository(&path, depth + 1).await?;
-                }
-            }
-            Ok(())
+        async move |dependencies| {
+            Ok(
+                move_to_repository(&postinstall_fetch_dir, &*JAVA_REPOSITORY_DIR, &dependencies)
+                    .await?,
+            )
        },
        &job.id,
        &job.workspace_id,
@@ -486,10 +481,122 @@ async fn install<'a>(
        is_sandboxing_enabled(),
        conn,
    )
-    .await?;
+    .await;
+
+    // coursier failing against the registry bails before the postinstall step ever runs, so the
+    // cache it was writing into has to be reclaimed on every path out
+    match remove_dir_all(&fetch_dir).await {
+        Err(e) if e.kind() != std::io::ErrorKind::NotFound => {
+            tracing::warn!("could not remove java fetch dir {fetch_dir}: {e}")
+        }
+        _ => {}
+    }
+    installed?;
    Ok(classpath)
 }

+/// Copies every fetched artifact out of coursier's cache into the `<group as path>/<artifact>/
+/// <version>` location the classpath is built from. Coursier's cache is laid out as
+/// `<scheme>/<host>/<registry url path>/<group as path>/...`, so the depth at which the maven
+/// layout starts varies with the registry url and only the coordinate can be matched on.
+async fn move_to_repository(
+    fetch_dir: &str,
+    repository_dir: &str,
+    deps: &[RequiredDependency<()>],
+) -> anyhow::Result<()> {
+    struct Wanted {
+        coordinate: Vec<String>,
+        destination: String,
+        display_name: String,
+        found: bool,
+    }
+
+    #[async_recursion]
+    async fn find_and_copy(
+        dir: &Path,
+        // components walked past below the fetch dir, compared against coordinates one component
+        // at a time so that windows' native separator cannot defeat the match
+        below: &mut Vec<String>,
+        wanted: &mut Vec<Wanted>,
+    ) -> anyhow::Result<()> {
+        if wanted.iter().all(|w| w.found) {
+            return Ok(());
+        }
+        let (mut subdirs, mut holds_artifact) = (vec![], false);
+        let mut entries = tokio::fs::read_dir(dir).await?;
+        while let Some(entry) = entries.next_entry().await? {
+            let name = entry.file_name().to_string_lossy().into_owned();
+            if entry.file_type().await?.is_dir() {
+                subdirs.push((name, entry));
+            } else {
+                // coursier's own bookkeeping (`.<name>.checked`, `.<name>.error`, checksums) is
+                // dot-prefixed; only a downloaded artifact is not
+                holds_artifact |= !name.starts_with('.');
+            }
+        }
+        // a repository coursier probed and did not get the artifact from keeps a directory at the
+        // coordinate holding nothing but that bookkeeping, and default repositories are probed
+        // before the configured ones, so claiming the first match would cache an empty directory
+        // as the installed artifact
+        if holds_artifact {
+            if let Some(w) = wanted
+                .iter_mut()
+                .find(|w| !w.found && below.ends_with(&w.coordinate))
+            {
+                copy_dir_recursively(dir, &PathBuf::from(&w.destination))?;
+                w.found = true;
+                return Ok(());
+            }
+        }
+        for (name, entry) in subdirs {
+            below.push(name);
+            find_and_copy(&entry.path(), below, wanted).await?;
+            below.pop();
+        }
+        Ok(())
+    }
+
+    let mut wanted = deps
+        .iter()
+        .map(|RequiredDependency { path, display_name, .. }| {
+            let suffix = path
+                .strip_prefix(repository_dir)
+                .filter(|suffix| suffix.starts_with('/'))
+                .ok_or_else(|| anyhow!("Internal Error: {path} is not under {repository_dir}"))?;
+            Ok(Wanted {
+                coordinate: suffix
+                    .split('/')
+                    .filter(|component| !component.is_empty())
+                    .map(str::to_owned)
+                    .collect(),
+                destination: path.clone(),
+                display_name: display_name.clone(),
+                found: false,
+            })
+        })
+        .collect::<anyhow::Result<Vec<_>>>()?;
+    // longest coordinate first: a group id ending in another one's coordinates (com.org.foo:bar
+    // over org.foo:bar) would otherwise be free to claim the shorter one's directory
+    wanted.sort_by_key(|w| std::cmp::Reverse(w.coordinate.len()));
+
+    find_and_copy(&PathBuf::from(fetch_dir), &mut vec![], &mut wanted).await?;
+
+    let missing = wanted
+        .iter()
+        .filter(|w| !w.found)
+        .map(|w| w.display_name.as_str())
+        .sorted()
+        .collect_vec();
+    if !missing.is_empty() {
+        bail!(
+            "the configured maven repositories did not serve: {}. \
+            Coursier reported success but no artifact for them was found in its cache.",
+            missing.join(", ")
+        );
+    }
+    Ok(())
+}
+
 async fn compile<'a>(
    JobHandlerInput {
        occupancy_metrics,
@@ -1012,3 +1119,161 @@ class Wmill {
    }
 }
 "#;
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn dep(
+        repository_dir: &str,
+        group_path: &str,
+        artifact: &str,
+        version: &str,
+    ) -> RequiredDependency<()> {
+        RequiredDependency {
+            path: format!("{repository_dir}/{group_path}/{artifact}/{version}"),
+            _s3_handle: format!("{}:{artifact}:{version}", group_path.replace("/", ".")),
+            display_name: format!("{artifact}:{version}"),
+            custom_payload: (),
+        }
+    }
+
+    async fn touch(path: &str) {
+        let path = PathBuf::from(path);
+        create_dir_all(path.parent().unwrap()).await.unwrap();
+        File::create(path).await.unwrap();
+    }
+
+    #[tokio::test]
+    async fn artifacts_are_found_whatever_the_registry_url_path_is() {
+        let tmp = tempfile::tempdir().unwrap();
+        let fetch_dir = tmp.path().join("fetch").to_str().unwrap().to_owned();
+        let repository_dir = tmp.path().join("repository").to_str().unwrap().to_owned();
+
+        // maven central has a single url segment before the maven layout, a nexus repository two,
+        // and a root-hosted mirror none at all
+        touch(&format!("{fetch_dir}/https/repo1.maven.org/maven2/commons-cli/commons-cli/1.4/commons-cli-1.4.jar")).await;
+        touch(&format!("{fetch_dir}/https/nexus.local/repository/maven-public/com/google/code/gson/gson/2.8.9/gson-2.8.9.jar")).await;
+        touch(&format!("{fetch_dir}/https/maven.local/org/apache/commons/commons-lang3/3.8.1/commons-lang3-3.8.1.jar")).await;
+
+        let deps = vec![
+            dep(&repository_dir, "commons-cli", "commons-cli", "1.4"),
+            dep(&repository_dir, "com/google/code/gson", "gson", "2.8.9"),
+            dep(
+                &repository_dir,
+                "org/apache/commons",
+                "commons-lang3",
+                "3.8.1",
+            ),
+        ];
+
+        move_to_repository(&fetch_dir, &repository_dir, &deps)
+            .await
+            .unwrap();
+
+        for (dep, jar) in deps.iter().zip([
+            "commons-cli-1.4.jar",
+            "gson-2.8.9.jar",
+            "commons-lang3-3.8.1.jar",
+        ]) {
+            assert!(
+                metadata(format!("{}/{jar}", dep.path)).await.is_ok(),
+                "{} was not copied to {}",
+                dep.display_name,
+                dep.path
+            );
+        }
+    }
+
+    #[tokio::test]
+    async fn the_empty_directory_a_404_leaves_behind_does_not_claim_the_coordinate() {
+        let tmp = tempfile::tempdir().unwrap();
+        let fetch_dir = tmp.path().join("fetch").to_str().unwrap().to_owned();
+        let repository_dir = tmp.path().join("repository").to_str().unwrap().to_owned();
+
+        // a repository that 404s is left with an empty directory at the coordinate. Each artifact
+        // is served by one of the two repositories and left empty under the other, so whichever
+        // one the walk reaches first, an empty directory precedes an artifact.
+        let central = format!("{fetch_dir}/https/repo1.maven.org/maven2");
+        let nexus = format!("{fetch_dir}/https/nexus.local/repository/maven-public");
+        touch(&format!("{central}/com/corp/public/1.0/public-1.0.jar")).await;
+        touch(&format!(
+            "{nexus}/com/corp/public/1.0/.public-1.0.jar.error"
+        ))
+        .await;
+        touch(&format!("{nexus}/com/corp/internal/1.0/internal-1.0.jar")).await;
+        touch(&format!(
+            "{central}/com/corp/internal/1.0/.internal-1.0.jar.error"
+        ))
+        .await;
+
+        let deps = vec![
+            dep(&repository_dir, "com/corp", "public", "1.0"),
+            dep(&repository_dir, "com/corp", "internal", "1.0"),
+        ];
+
+        move_to_repository(&fetch_dir, &repository_dir, &deps)
+            .await
+            .unwrap();
+
+        for (dep, jar) in deps.iter().zip(["public-1.0.jar", "internal-1.0.jar"]) {
+            assert!(
+                metadata(format!("{}/{jar}", dep.path)).await.is_ok(),
+                "{} was not copied to {}",
+                dep.display_name,
+                dep.path
+            );
+        }
+    }
+
+    #[tokio::test]
+    async fn a_group_id_ending_in_another_coordinate_does_not_claim_it() {
+        let tmp = tempfile::tempdir().unwrap();
+        let fetch_dir = tmp.path().join("fetch").to_str().unwrap().to_owned();
+        let repository_dir = tmp.path().join("repository").to_str().unwrap().to_owned();
+        let root = format!("{fetch_dir}/https/nexus.local/repository/maven-public");
+
+        touch(&format!("{root}/org/foo/bar/1/bar-1.jar")).await;
+        touch(&format!("{root}/com/org/foo/bar/1/bar-1.jar")).await;
+
+        let deps = vec![
+            dep(&repository_dir, "org/foo", "bar", "1"),
+            dep(&repository_dir, "com/org/foo", "bar", "1"),
+        ];
+
+        move_to_repository(&fetch_dir, &repository_dir, &deps)
+            .await
+            .unwrap();
+
+        for dep in &deps {
+            assert!(
+                metadata(format!("{}/bar-1.jar", dep.path)).await.is_ok(),
+                "{} was not copied to {}",
+                dep.display_name,
+                dep.path
+            );
+        }
+    }
+
+    #[tokio::test]
+    async fn artifact_missing_from_the_cache_is_named_in_the_error() {
+        let tmp = tempfile::tempdir().unwrap();
+        let fetch_dir = tmp.path().join("fetch").to_str().unwrap().to_owned();
+        let repository_dir = tmp.path().join("repository").to_str().unwrap().to_owned();
+
+        touch(&format!("{fetch_dir}/https/nexus.local/repository/maven-public/commons-cli/commons-cli/1.4/commons-cli-1.4.jar")).await;
+
+        let deps = vec![
+            dep(&repository_dir, "commons-cli", "commons-cli", "1.4"),
+            dep(&repository_dir, "com/google/code/gson", "gson", "2.8.9"),
+        ];
+
+        let err = move_to_repository(&fetch_dir, &repository_dir, &deps)
+            .await
+            .unwrap_err()
+            .to_string();
+
+        assert!(err.contains("gson:2.8.9"), "unexpected error: {err}");
+        assert!(!err.contains("commons-cli:1.4"), "unexpected error: {err}");
+    }
+}