diff --git a/.github/den1420_post_patch.py b/.github/den1420_post_patch.py deleted file mode 100644 index 07bcab35..00000000 --- a/.github/den1420_post_patch.py +++ /dev/null @@ -1,254 +0,0 @@ -from pathlib import Path - -path = Path('src/mise.rs') -text = path.read_text() - -old = '''#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq)] -struct MiseConfig { - #[serde(default)] - tools: BTreeMap, - #[serde(default)] - env: BTreeMap, - #[serde(default)] - tasks: BTreeMap, - #[serde(default)] - settings: BTreeMap, -} -''' -new = '''#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq)] -struct MiseConfig { - #[serde(default)] - tools: BTreeMap, - #[serde(default)] - env: BTreeMap, - #[serde(default)] - vars: BTreeMap, - #[serde(default)] - tasks: BTreeMap, - #[serde(default)] - settings: BTreeMap, - /// Manager-native sections that Zed does not execute yet. They remain - /// lossless and inert instead of being silently discarded. - #[serde(flatten, default)] - unknown: BTreeMap, -} -''' -assert old in text, 'MiseConfig block not found' -text = text.replace(old, new, 1) - -old = ''' Self::Versions(values) => { - values.sort(); - values.dedup(); - } - Self::Options(options) => { - if let Some(version) = &mut options.version { - let mut values = version.as_vec(); - values.sort(); - values.dedup(); - options.version = Some(OneOrManyString::Many(values)); - } -''' -new = ''' Self::Versions(values) => dedup_preserving_order(values), - Self::Options(options) => { - if let Some(version) = &mut options.version { - let mut values = version.as_vec(); - dedup_preserving_order(&mut values); - options.version = Some(OneOrManyString::Many(values)); - } -''' -assert old in text, 'version normalization block not found' -text = text.replace(old, new, 1) - -old = ''' plan.extensions.insert( - "mise.env".to_string(), - serde_json::to_value(&config.env).context("failed to encode mise env")?, - ); - plan.extensions.insert( - "mise.tasks".to_string(), - serde_json::to_value(&config.tasks).context("failed to encode mise tasks")?, - ); -''' -new = ''' plan.extensions.insert( - "mise.env".to_string(), - serde_json::to_value(&config.env).context("failed to encode mise env")?, - ); - plan.extensions.insert( - "mise.vars".to_string(), - serde_json::to_value(&config.vars).context("failed to encode mise vars")?, - ); - plan.extensions.insert( - "mise.tasks".to_string(), - serde_json::to_value(&config.tasks).context("failed to encode mise tasks")?, - ); -''' -assert old in text, 'env/tasks extension block not found' -text = text.replace(old, new, 1) - -old = ''' plan.extensions.insert( - "mise.settings".to_string(), - serde_json::to_value(&config.settings).context("failed to encode mise settings")?, - ); - plan.extensions.insert( - "mise.source_sha256".to_string(), -''' -new = ''' plan.extensions.insert( - "mise.settings".to_string(), - serde_json::to_value(&config.settings).context("failed to encode mise settings")?, - ); - plan.extensions.insert( - "mise.unknown_top_level".to_string(), - serde_json::to_value(&config.unknown) - .context("failed to encode unsupported mise sections")?, - ); - plan.extensions.insert( - "mise.source_sha256".to_string(), -''' -assert old in text, 'settings/source extension block not found' -text = text.replace(old, new, 1) - -old = '''fn ensure_project_local_path(path: &str, field: &str) -> Result<()> { - let path = path.trim(); - ensure!(!path.is_empty(), "{field} cannot be empty"); - let parsed = Path::new(path); - ensure!(!parsed.is_absolute(), "{field} must be project-relative"); - ensure!( - parsed - .components() - .all(|component| !matches!(component, Component::ParentDir)), - "{field} cannot escape the project root" - ); - Ok(()) -} -''' -new = '''fn ensure_project_local_path(path: &str, field: &str) -> Result<()> { - let path = path.trim(); - ensure!(!path.is_empty(), "{field} cannot be empty"); - let parsed = Path::new(path); - let has_windows_drive = path.as_bytes().get(1).is_some_and(|value| *value == b':') - && path - .as_bytes() - .first() - .is_some_and(|value| value.is_ascii_alphabetic()); - let has_parent = path.split(['/', '\\']).any(|part| part == ".."); - ensure!( - !parsed.is_absolute() - && !has_windows_drive - && !path.starts_with('~') - && !path.starts_with("$HOME") - && !path.starts_with("${HOME}") - && !path.starts_with("%USERPROFILE%") - && !path.starts_with("\\\\") - && !path.starts_with("//"), - "{field} must be project-relative on every supported platform" - ); - ensure!(!has_parent, "{field} cannot escape the project root"); - ensure!( - parsed - .components() - .all(|component| !matches!(component, Component::ParentDir)), - "{field} cannot escape the project root" - ); - Ok(()) -} - -fn dedup_preserving_order(values: &mut Vec) { - let mut seen = BTreeSet::new(); - values.retain(|value| seen.insert(value.clone())); -} -''' -assert old in text, 'path validation block not found' -text = text.replace(old, new, 1) -path.write_text(text) - -test_path = Path('tests/mise_adapter.rs') -tests = test_path.read_text() -insert = r''' - -#[test] -fn multi_version_order_is_semantic_and_preserved() { - let root = temp_project("version-order"); - write( - &root.join(".mise.toml"), - r#" -[tools] -node = ["22", "20", "22"] -"#, - ); - - let imported = import_mise(MiseAdapter::discover(&root).unwrap()).unwrap(); - assert!(imported.source_document.contains("[\"22\", \"20\"]")); - assert!(!imported.source_document.contains("[\"20\", \"22\"]")); - fs::remove_dir_all(root).unwrap(); -} - -#[test] -fn unknown_top_level_sections_remain_lossless_and_inert() { - let root = temp_project("unknown-sections"); - write( - &root.join(".mise.toml"), - r#" -[tools] -node = "22" - -[hooks] -postinstall = "echo never-run-by-import" - -[plugins] -custom = "https://example.invalid/plugin.git" -"#, - ); - - let imported = import_mise(MiseAdapter::discover(&root).unwrap()).unwrap(); - let unknown = imported - .plan - .extensions - .get("mise.unknown_top_level") - .expect("unknown sections are represented"); - assert_eq!(unknown["hooks"]["postinstall"], "echo never-run-by-import"); - assert_eq!( - unknown["plugins"]["custom"], - "https://example.invalid/plugin.git" - ); - assert!(imported.source_document.contains("postinstall")); - fs::remove_dir_all(root).unwrap(); -} - -#[test] -fn vars_are_preserved_for_future_native_template_evaluation() { - let root = temp_project("vars"); - write( - &root.join(".mise.toml"), - r#" -[vars] -channel = "stable" -retries = 3 -"#, - ); - - let imported = import_mise(MiseAdapter::discover(&root).unwrap()).unwrap(); - let vars = imported.plan.extensions.get("mise.vars").unwrap(); - assert_eq!(vars["channel"], "stable"); - assert_eq!(vars["retries"], 3); - fs::remove_dir_all(root).unwrap(); -} - -#[test] -fn portable_path_validation_rejects_windows_and_unc_absolute_paths() { - let root = temp_project("portable-paths"); - for path in [r"C:\Users\alex\mise.toml", r"\\server\share\mise.toml"] { - let adapter = MiseAdapter::for_source(&root, path).unwrap(); - let error = import_mise(adapter).unwrap_err().to_string(); - assert!(error.contains("project-relative")); - } - fs::remove_dir_all(root).unwrap(); -} -''' -assert 'fn multi_version_order_is_semantic_and_preserved' not in tests -test_path.write_text(tests + insert) - -doc_path = Path('docs/mise.md') -doc = doc_path.read_text() -needle = '- Project-local config only; global/user configuration is ignored.\n' -replacement = '''- Project-local config only; global/user configuration is ignored.\n- Unknown top-level mise sections are preserved losslessly under `mise.unknown_top_level` and remain inert; import never executes hooks or plugins.\n- Multi-version order is preserved because PATH/default selection can be order-sensitive.\n''' -assert needle in doc -doc_path.write_text(doc.replace(needle, replacement, 1)) diff --git a/.github/workflows/apply-den-1420-mise.yml b/.github/workflows/apply-den-1420-mise.yml deleted file mode 100644 index fae8ceb5..00000000 --- a/.github/workflows/apply-den-1420-mise.yml +++ /dev/null @@ -1,112 +0,0 @@ -name: apply DEN-1420 mise adapter - -on: - push: - branches: - - feat/den-1420-mise-dev-adapter - -permissions: - contents: write - -jobs: - apply: - if: github.actor != 'github-actions[bot]' - runs-on: ubuntu-latest - steps: - - uses: actions/checkout@v4 - with: - fetch-depth: 0 - - - uses: dtolnay/rust-toolchain@stable - - - name: Apply reviewed mise adapter patch - shell: bash - run: | - set -euo pipefail - base64 -d <<'PATCH_GZ' | gzip -d > "$RUNNER_TEMP/den-1420.patch" - H4sIAAMqcGoC/+08a3PbtrLf9SsQpjeWKpKybOclNzl1HCX1aWJ7bCc9bZqhKAqyGFOkyocdNc5/v7sLgAQfkpW2d+bOneuZ2CKx - WCz2vQsoE386ZZZ16afM7SWx15vwaztO2Fh7aFmWVRptdbvdMsSPPzJr56H5kHXh9yP2448t5kVhkrKXw1cH795cOO+Hx+8H7EGS - xuwZM+w/+QSn9hbLdBaFvWseXhv7as7x0X+cs+Hw+OLsV6c07bfhS+fl8L2DAAeHF0fvhzCpKya9PTofrp9FEPm0nL73zuEJTIKB - fIofekE24Q483Wsbtt2zgVbLC3xrGriXiZ1G88DoAI4Wu/9hwmP/mrdf8nF2abLDIAo5/IkWS5OdunHqu8HwD5Phv/dukPFhmM07 - H4lhu7vAqu7untkXHMOfXq/Hjvk1j1nMLR6m8CGdxVF2OWOj0P/MgBAeRIuRLcAJ1CymnvE/Mj/mzGUhd+PxkgHBV/AYTuDVTRRf - +eElIRox/pl7WeqOA263ujhfzp2Yre7XVrfV/atb6y6yMePwmb3k12/9hL+NJpx9EYsgkYfRfBElnN346Ywt4ugT91IriDw3YHMA - R8lM/cssdlM/CtnNjIfivZ8w99r1A43mgyyNzALzKs7RdNzwSM6TfOtW+XYnNRVOIkSZlazMSvYVtASF/egJCvvxU7O/nQv7/gc3 - vmwDW4G718hBB9lmAvOuK+pumCD5qZsFqUOAOOzC3g3gN6FCnoNYB8jzY/8zshwWLzb4y4ynM2RJBBtayX7aD6n3iEUxG9nFs2Tc - hiSjra2juatoxgUGuqKAUAp1fim0fQ7iREqnfsD32cj1R8ydTBKQL2fRIrX8kB0cwdaigCUzf84mwHsvjeKl/U18Pj07eXX0ppFu - +aLEbkkQUX8qPpvCEfafmHvgCcG+H+ey9qcs4Ck7j+a87cE+O4B27i7H3Ik5aasjtdUBMbYfxFGUmsybAsUPYIugeknnX8qI8Cfm - aRaH7ORKYNsXI1/Fn80XQ/6vXo2tXI2J1YRuP9x5iht+uPfYLDwZ/tjId8OyYEtGp/o6RBuE190qNNJUe48vnTks7oQu7EpSSQra - 6dRXlMKprSrfl5HIl4CHdvMUQ1j3Uf+h2X/yMN/PMXg/Yc7dabiOm4JwZCkElFM3nUk/A+yFF4fkTuQrSQC8lpp+EF8mMNZh1nNw - Igmo3A8nBPODv7vz/LlSAJCvzgD27JluQoMBubeCgbe3qOqDwbUbO1HSrobKju0nToK60lk5x3BE/Hx59OqVUZvQqJfIME0vxQdN - LZEPpJgoWfHYRq51GA8SruO8a7vK3epz8GcMweJeu/wOf6SOAa1y3o2b0ANPUnDabJylLIw29400fxplEBv8MPEh2pFnugkxRngz - 7l1FgHEKs758Ncw6Pbhre+Ini8Bd6jLAH8VBzbrXsHlf8RlYdg/ov4zduZMHzbYhTOuf4+0mnJRBEqJ3GKVFBGcQTU8PLn4yVuyQ - L2I/TIOwKkADckeVBg0k1798rS6VhflC+xjCUz/MUBoU9FA6IcTza85+A2Sg534chRhlqtIRatkkG53qDXQedOgKfYzSd/XcfiAN - Yb9m2dM4+hNSnwcP2D0FjXZHs0oibFJzEIycTzv+8lWJSERNd/LJ9TCqkvRy4urqWWVAZTgnbA2HSnwoUiXgBDkYL4tjoMSBkXbH - RlHxz2nbQKNLUWJIbpyFZEso+gKD0fnXfoF5DvKH1GaOGdozzDHx02AQ8hul9hJYAlXjDuJtika1dwUBNWiplk1oRAjs/p0QuBq8 - iHbdb4l2dTxUjllUjtVw0ZiDYw1xWBttQtyIUaUaNs3SBuGxFqRMZvSNKhClbqc/vwbA10fnAAm52wOIs3bML32o3pYrJ/x08nao - gGcQj1YCHry7+Ml5d/ZGAUMGO3OyONAnKBWGvFNEsH1d5cfootx4qUxfhhVHvc+n1DI37geg9RgjFay9cHGlsv1LhRZ0E98Oh0dv - jo5fO+hfz4FyhalqkXVvo6FFOoT548aACApTnyI/bBcFPO6oJ6BKjhwnJ0sIAnNnMxwC2GpANU3AScTcTTnicNwgUE4Tnyvh0kZ3 - 5ygvAkkMRN25m0KkIhToRPwkCgDZRC/vyPNpuy38WSd3MqtoqW2zUyOiEjsVTQ2ZyQoqxRo6sfW5NTpWphRavC3tTpM4egg0cBCZ - eFNhc+FWqgLIPXhD9byAbJjNXMwEGAUcxGCUaKh558IkSbXfnBz+PHxpVN1BBerw5PjV0WvIVs9KYl094fzX84vh2/K8umBXTj95 - Pzw7O3qZL4x15PHB2+F5sbzi6AZILk5O3jjwcH50cnxewmWEkGAYTXace400uuIh+gxyV/S00lkoJ3dx8vPwGNCLuatrSfR7CnOS - LdyxC/4MXt65wPm704MXB7BH4UkRz+pVrviyvgq83HyVn4cYCRDPOk4lMx6I/SgnSG+alpFhjMah/MFHMXv1LkRELKEXr9bgFwBy - AYmgtoLCFkYOVBupu5biMLIkkLEaU933V7AIgDUYAH9044wzP5isQUNQFkGtwaU8n4NxeQ02BWch3Bp8VEc42OtZw3iEsShJWS3R - xIv9RVqSqEo4V2L2lLaIuU0pMUgnzRJytGKmeNFeG0aaQ4hB7QgMH+R8J1r7TCtyGIhzvkGur6XykuzcU0OpI1hCpNrYFAJ6s/Am - dhdOFLf7HcwCaZvTsFbhy74IdTlkewOfX2TTvMPxzYkTVtcecM+fgE5AGS4yDTf0oBKN4qScMSFwHuEA9oORF/PoZIvS3vhYjd1F - jESJqQVFSpO7+P3yHFAiydvG+q1SRurtkQqir42VMqLPd/6sSBZrJTvkFlf1Ylv+of5WWWSNDNeFJ6WmFqqyXEuRp344ad/mZNZS - og+GfemnxPzZJf359Mn4WOeR7YOCV3MOUVGEy/bt3I2veHxbFYx43bH5Z6gKgLBSBqRRWSgwqZY2kkYOJi/g4qbtqmbntbwQ2hoG - ofLkzemG1CpP70ucC3DLt7j8LbuHf2w3caIEj6fa1Irj80W6LO2q2ImDzTR0HkiKLIdto6M0q2iVi6ScNB83ZFR3WdR8+KnUs6et - PthCZ+B7DM/Q5H7BRXkzNtfOf/Cn1F/CExz27Lk8GjBXQImDHQQTNfAquLxbhaCqKaWgv4odMdhRqTBuaOQ37YgVO5Iz9A55MXcw - eClOC4gIdXIgjgb620+28ZS0v/P4aaVXrsWOARmjWR6jfKNxpHLuI7lVce61g5YcrISrgRf5fqrL5vnHgKVxJg8/+v2dbTwM6O/C - 3/oOZW8KuyFmw0hNupWWbeOQJuemtVRzpGnU9ZVg+rvb5i6SvbsD5FfJRsOVLfq8sySf2xDeE0wKvMBX+cB+ebabJDyGFOSPewRs - w+7NusCqzr46C/dvNqt7544FJQvMklQPfJhGe9/Z2aG97+08NJ80isy+5KkDqDLMH9Czs2b/CyC3DH4RPB64QaSDiEQhzaAWmDZV - 37CguJrPyASmwdtX6en+BXpEk63csRT8eEznWf29J4/N3eIID89+5HliCiFORqdyfgNO2V0mjjgGcNIZd8QxgHPtJY46CiiHf9Qt - qDMXoEL4RwgHP2E7SSVV7WpzJT9WELNsjAsALby4Gqy1ZELqy6OySgg5wV0skt4NH9/VeKlMo5jdaaSxqVMiVl8JfhNDcG/Xd1Nk - Yx1IDD7gOW/y8XcskzREVVUi5dcTT7W6yVSTvmn16gY3Wrth5brzrdFRBhFZ3+rVK431DWj/q6Q3M62JPp28WmnTaCQqVXIga1Fn - EA5k45BiilKHTGgBkQ5sppK0r+dunoRpeU7MF1FPJ7LCc5mdDQZYEOngKgFa7am+mQz7n6SjxmSmmOyG7iWUzlq151xxvkicwA0v - MxhDBwU2CQk9j51JlDp/8gnxeaLfBGu48MTGTW/l3bCmEXFJrGkEnezjPXSx8LvfJwc7UTmTzARbDPId9IkGREl5Z8WnJqJIPxHI - ZJQvmHgeq5IA28AT+g8yECe2DCG2WB65CirZdFul1XUD3004VuKiHsT6I10u6AIIpICgkQBUo7Or6JT9zm8gFHLRVYTKiP2xLJZJ - 5CU9BTifAHsrb6Q4qm+FKKpvRayjyI+/UQoXM54fovkJCCXwxzwGhQmWzGXff+/FUZJYSpdKTQUviuKJH7pQTHz/vSnOelsW6HIA - Vi7aDrD7QwjGyI7F3GQL+p1dm+x1hMzKAp6Y7FWQpSny6hizJJjyG59sYcc4nrugbgvXu8KlY55EAfDUbnX/xhoipyqt1OqKpVh9 - Jbrcs40a28U/Ik3csdlRyEaU05KwRyR4s3zPio6cOVR3ScpGmAD26GaeTVfxopgMeKS9Gi/za2ziELTpDiB7B+jlyqRgI7zfhRkI - 4ZNH3HhZQcAo1SOwqesHkLMChRO7Ze0CshB4kxBs3jSks/QFUA2UmMgJkC5XnElysmh3gmvAw1YXsB1M8QYeJGHIXbmXeGkiejIU - wSnJo+SOy2jk7KqXLppu9ymO0KuCJePlAty2aIpF4mLMqHJzgeDkA8NopKhpuACI2y1dn9z7J7knyszQnwJ8Lz+alyhBDdlBPkxL - IgguJPFPmGznIqVh1BJHQyEsgqXsyP5zcXUpOQiTMHNIeQgaAHs44xlKJmTUJ6ELjnQ2OyLrENkcYqhf3h2R4bmgmimwDFXg4T+K - jqKk6J5LsdjsMAKNigJgny/1Z1Q6wSYVuyUluM2lLIrAHBBIQTaNxAn1CNgAdL+gNjZKRYbTkp9DeO7TZUpxy9QFzoPofHxwPTI5 - qtUhyD/6B5FRgy1DT+P1LEs6acFKPofo8O/zk+OCDaqfPRJe6xG5q+6jXXOXHL11/z7ZJa6PLaDwutUVr5RP1Eag6im5N0wexTWS - yu0Z/faTcnbk07YwlFyLndgyysyQL354Tfc8QK4gRGrTTWhpcBi4WbWfwsOZaFAgLNSnFoB5WZzAdJsNlZLhvq7d2EfbTKD4Q9sB - BkOGPNmXwYR/XgS+B0yTMmkxFdJ6ngtJbmV+ileQI1g9cP0JqWaa70DtqWWNtItJIzaJuLjwpHyKYKbgqxpLoiz20AZgJPbAINOE - B1Ow8JY4amJuAOowWQrl4JPcZUhk6BL1iCBitrAZBQuEHR3Tlfnzn4Zv3oyYjO2CNktitqEqArpSxCpaBeISPDIULG6SeSQkVAcL - xyGI+uh63AkKDNRijp0ycFBLcIdlHx9E0VXCohBSCGTd5rfq6JgCON3qAlfEnpRPzhboroFd7LUPrvUtj0EPfDcgc/h39ilLk6wo - kfPbE+woxXRBbDppiCCQaTFNjMy27ZFM3nJoV8SR1E2uTJAg5R+Y8mFLF1/SoXaCLWjBYtRQCCkueukQqFZ6AtFKMwSl/ZCG1tVf - D2fSXxUWYWMxgEoPODjeWKZrA1M/TsBJ/oKqi04lJxFdS4jCRGEBHV6AhkohzyS9BlStrqCKYld8LS+N5ff6BAOorYvUqMv5gbsk - JdW8AOHSPEQ9ZWGarhRuOjcyoGDKY8QaglsndARObXDc+C+SRSLojUxWS8ObL76N8gJrhBfBydgF7i0RUnGblMf5lxCccXBkYw0N - kYI8TiJjB9BGPXSZXlwG0Rj0GkdKlzZU/kAqQA37VrdsCqoV73O8vgjlmRYb6V4j6otIKND/ARDIxhLrgRJmINQ5MAa4LyzHm4FP - Q/t184zgEneOGfIRhZKbGWYWAXevEUyk7/ESE+BQ3R6UhgDE6CzFz7AdvMVHKgqpCRqYVIUxnwHGKCb5HLBmTwYJrpa4NTkuJoWh - qQ+UX1gOieOcVY5sYz8M2O50xGxDPwzacZcjZpv6YaRrA0fM7vLDELdPQaX4hIceGuhAXtd/TF/UevrYfKgKmCJBUyXx9yPWE094 - sQKe9DBPX4tI9lsM8nwqV60dzDXoni1mLaqYTyguHhXuIFm4N6Fw4x6krmaeg+QRWvK7iL8xn7t+WFwbNlsWhWNMPURIXgrLhuqW - usFFlJB2Rd8LyRGCUt5BkJ4B3UUXeOucMPY36RLl+CPRiX8iy0vIyu4TPRYGkMINah4drWw0GiUz6RplpdDqSjtgDd42v69teWyL - 3IJF0XKLUCFGce2M4q7YBzgW5JT0M4mpa4RgFnUYKcVStQg5TSo4lKP9LMo9tFFHluM9e+yrRDy6CeVq5S9iiUqr1c07D5SmJTZ7 - Rd6Mim1MgjWvLr6qNZlowQEdErsv8lzFT42PmOcSH0kSe09JEk/UcdAtIiItGOFnZSjSmG8lABXv2jDYOw52b1UYL41i46mYK3s+ - JQj5nSQEKjfo4M8sG/fwm2jTILrJGzuCRHspGnZ3Q6kG3gaQsqG3ASQxkE7TqNgQzYg5H7AXbgJZxm/465WPv2VRh9WXSS2JLEbd - QtmdXvzKknl0xUWrcwUCYasboUGq9nZRqvC7fsZ3CarArD8wBdwqFZJbzPgOm3zGOvjSbaRiRnf1DApvG6D2TJbXewW8PLSkc6r+ - nvmkftSKpo0Xcz5vsd/LI7f5AkmajS0CUl8W1HrZYJbPjO/O3h0fD8+ci+Hb055wRH5Y2tf8Cq/2WgsgTU0rjUN9x55rg4TFYD/8 - sIXqv6WD3r/Xy5K4h0AYT8YuOjXtCJynzOJZBL5owVHK+qA/ZR8+wDJf+gPrq4GnfHTvnHI1g338uE/+uSwQKpOnbOu/kt/DLbZF - Pm9ne+eR/cTetpA3W2V4iJMp29bfTX39CZvxGgmU25Q2MPOn6boJlrUReMNXmiWCfiNLCG74n9Ph4YXz6uzkt+HxwNoWTOqvYI1a - Srt027SEBLTCHLa4Qovw62Frt25XTFHwK6/ZStLy4nETHCtu2UpUIV2HWiVojGOx/NL5+cW7F1IIz/plICgoje9+LO0Ip5SsYwZh - kHU/1wxEP+NjpftYTUYph9YaZhNMxQK+fN0qbFXC92TD0v6UROG6yfKIET6FWBqCAe7s2Hv2ttGAszgNKzFjso5W/DbDs4JNg+/o - 62MwA83hxdGxQdnO72XZ6/3YhrE8MaoNlfOk2rDeaFyBQaSMxndtdaJiCZfWMWqw4KqlklZViixuX6lw0/9lIACkLEgZoqut6gqF - y5cQZfUCT1H3Ec/6/xMsl4Xl/2JR4GUm0tialWWh6AzYeLeC7ayH4XFcUt+6j+XeLGJbkiGiGNaqrCwEL0MVJB6+eR5fYA1B3adE - XfGlVBfM68FOQ6Dqr/ZfShm2NvhW4NYGu9RxD4B3OW1V1/f/Gra5sevfd6kbudjlBraeA5bFFEbOmqAC/N0krlSQbBJaKlNWRxcK - BusWAKf1fzokrPZD+R2CNX4oh9nUD+Ub29ATKT9UPif9O87oW75zv7XBnvWqR35BtVnbpeRyJftvfe+pTBZJAAA= - PATCH_GZ - printf '%s %s\n' 'd0d0a867d249556a41322169483f27f9107a58d4131a2a92270e16d4967623bc' "$RUNNER_TEMP/den-1420.patch" | sha256sum -c - - git apply --check "$RUNNER_TEMP/den-1420.patch" - git apply "$RUNNER_TEMP/den-1420.patch" - rm .github/workflows/apply-den-1420-mise.yml - cargo fmt --all - git diff --check - git config user.name 'github-actions[bot]' - git config user.email '41898282+github-actions[bot]@users.noreply.github.com' - git add -A - git commit -m 'feat(DEN-1420): compose zed dev with mise' - git push origin "HEAD:${GITHUB_REF_NAME}" diff --git a/.github/workflows/kick-den-1420-mise.yml b/.github/workflows/kick-den-1420-mise.yml deleted file mode 100644 index b6a2ba74..00000000 --- a/.github/workflows/kick-den-1420-mise.yml +++ /dev/null @@ -1,125 +0,0 @@ -name: kick DEN-1420 mise adapter - -on: - pull_request: - types: [opened, reopened, synchronize] - -permissions: - contents: write - -jobs: - apply: - if: github.event.pull_request.head.ref == 'feat/den-1420-mise-dev-adapter' - runs-on: ubuntu-latest - steps: - - uses: actions/checkout@v4 - with: - ref: feat/den-1420-mise-dev-adapter - fetch-depth: 0 - - - uses: dtolnay/rust-toolchain@stable - with: - components: rustfmt - - - name: Extract, verify, and apply reviewed patch - shell: bash - run: | - set -euo pipefail - python3 - <<'PY' - import base64 - import gzip - from pathlib import Path - - source = Path('.github/workflows/apply-den-1420-mise.yml').read_text(encoding='utf-8') - lines = source.splitlines() - start = next(i for i, line in enumerate(lines) if "base64 -d <<'PATCH_GZ'" in line) + 1 - end = next(i for i in range(start, len(lines)) if lines[i].strip() == 'PATCH_GZ') - payload = ''.join(line.strip() for line in lines[start:end]) - Path('/tmp/den-1420.patch').write_bytes(gzip.decompress(base64.b64decode(payload))) - PY - - printf '%s %s\n' 'd0d0a867d249556a41322169483f27f9107a58d4131a2a92270e16d4967623bc' /tmp/den-1420.patch | sha256sum -c - - patch --batch --forward --fuzz=3 -p1 < /tmp/den-1420.patch - - # One low-context assertion hunk can land beside the contract-parity - # test under GNU patch fuzz. Relocate it structurally, then verify - # every product file against the reviewed pre-format blob. - python3 - <<'PY' - from pathlib import Path - - path = Path('src/dev.rs') - lines = path.read_text(encoding='utf-8').splitlines(keepends=True) - - def assertion_bounds(index: int) -> tuple[int, int]: - start = index - while start >= 0 and lines[start].strip() != 'assert!(': - start -= 1 - if start < 0: - raise SystemExit('could not locate assertion start') - end = index - while end < len(lines) and lines[end].strip() != ');': - end += 1 - if end == len(lines): - raise SystemExit('could not locate assertion end') - return start, end + 1 - - help_index = next( - i for i, line in enumerate(lines) - if 'fn augmented_root_help_contains_the_command_and_alias()' in line - ) - - misplaced = [ - i for i, line in enumerate(lines[:help_index]) - if 'Some("mise")' in line - ] - for index in reversed(misplaced): - start, end = assertion_bounds(index) - del lines[start:end] - - help_index = next( - i for i, line in enumerate(lines) - if 'fn augmented_root_help_contains_the_command_and_alias()' in line - ) - next_test = next( - (i for i in range(help_index + 1, len(lines)) if lines[i].strip() == '#[test]'), - len(lines), - ) - if not any('Some("mise")' in line for line in lines[help_index:next_test]): - nix_index = next( - i for i in range(help_index, next_test) - if 'Some("nix")' in lines[i] - ) - _, insert_at = assertion_bounds(nix_index) - lines[insert_at:insert_at] = [ - ' assert!(\n', - ' develop\n', - ' .get_arguments()\n', - ' .any(|arg| arg.get_long() == Some("mise"))\n', - ' );\n', - ] - - path.write_text(''.join(lines), encoding='utf-8') - PY - - if find . -name '*.rej' -print -quit | grep -q .; then - echo 'patch left reject files:' >&2 - find . -name '*.rej' -print -exec cat {} \; >&2 - exit 1 - fi - - test "$(git hash-object src/dev.rs)" = 'ad55ef52dc4703afa7025bcba3c60de5d444cca1' - test "$(git hash-object .dev-cli-flags.toml)" = 'd4b6a979f3274a87370ef518d58a49705fcdc3af' - test "$(git hash-object docs/develop.md)" = 'a53eb8e6f35aaf84465149d1146d65166e6d12ff' - test "$(git hash-object .github/workflows/develop-shell.yml)" = '14c23135d931e20423084be1a698d67ac155e15e' - - rm .github/workflows/apply-den-1420-mise.yml - rm .github/workflows/kick-den-1420-mise.yml - cargo fmt --all - git diff --check - cargo test --locked - - git config user.name 'github-actions[bot]' - git config user.email '41898282+github-actions[bot]@users.noreply.github.com' - git add -A - git commit -m 'feat(DEN-1420): compose zed dev with mise' - git push origin HEAD:feat/den-1420-mise-dev-adapter diff --git a/docs/recursive-installs.md b/docs/recursive-installs.md new file mode 100644 index 00000000..c0d67d8c --- /dev/null +++ b/docs/recursive-installs.md @@ -0,0 +1,64 @@ +# Recursive and concurrent installs + +`zed install` resolves dependencies from every installed package manifest, not +only the consumer's direct dependency table. The resolver keeps one selected +version per `org/name`, detects incompatible requirements, terminates cycles, +and deduplicates diamond-shaped graphs before the project tree is changed. + +## Concurrency + +Artifact download, integrity verification, and extraction run through a bounded +worker queue. The default is five concurrent artifact installs: + +```text +ZED_PKG_INSTALL_CONCURRENCY=5 zed install +``` + +`ZED_PKG_INSTALL_CONCURRENCY` may be set from 1 through 32. Invalid values and +zero use the default. Registry metadata resolution stays deterministic on the +coordinator thread; only independent artifact work is concurrent. + +## Cross-process locking + +Each content hash owns a lock file under: + +```text +$ZED_PKG_HOME/locks/artifact-.lock +``` + +The lock uses the operating system's blocking file-lock primitive. A waiter +sleeps in the kernel until the owner exits or releases the descriptor; there is +no retry timer or filesystem polling loop. The lock covers download, SHA-256 +verification, atomic publication into the download cache, and extraction into +the content-addressed store. Two terminals or CI jobs sharing a Zed home can +therefore install unrelated artifacts concurrently while downloading any one +artifact at most once. + +Recursive prefetch and the transactional installer use the same artifact +acquisition function. Frozen installs, build preparation, `zed add`, and +`zed remove` therefore cannot bypass the per-hash lock through an older direct +cache-download path. + +Downloads are written to a temporary file in the cache directory, verified, +and atomically renamed. A corrupt cache file left by an interrupted older +client is removed and fetched again while the artifact lock is held. + +## Materialization + +The concurrent phase only warms the global store. The existing project +transaction remains responsible for lockfile writes, adapter wiring, rollback, +and dependency materialization. Unix installations therefore continue to use +store-backed symlinks by default; explicit copy mode and the non-Unix fallback +remain self-contained copies. + +## Validation + +The unit suite races the recursive and transactional acquisition paths against +the same absent content hash. The first download is held open while the second +path attempts acquisition; the test proves that the second path never enters +its registry download and that the aggregate download count remains one. + +The companion end-to-end suite also runs multiple independent CLI processes +against one shared Zed home, verifies the complete transitive lock graph, and +checks that project packages are store-backed symlinks rather than copied +package trees. diff --git a/src/config.rs b/src/config.rs index 34bf3811..89210180 100644 --- a/src/config.rs +++ b/src/config.rs @@ -156,6 +156,7 @@ struct ManifestOverride { thread_local! { static MANIFEST_OVERRIDE: RefCell> = const { RefCell::new(None) }; + static INSTALL_PREFETCH_CONFIG: RefCell> = const { RefCell::new(None) }; } struct ManifestOverrideGuard; @@ -168,10 +169,44 @@ impl Drop for ManifestOverrideGuard { } } +struct InstallPrefetchGuard; + +impl Drop for InstallPrefetchGuard { + fn drop(&mut self) { + INSTALL_PREFETCH_CONFIG.with(|slot| { + slot.borrow_mut().take(); + }); + } +} + fn normalized_project(project: &Path) -> PathBuf { fs::canonicalize(project).unwrap_or_else(|_| project.to_path_buf()) } +/// Mark one facade operation so an in-memory manifest proposed by `zed add` +/// or `zed remove` is recursively prefetched before the implementation-local +/// reinstall begins. The context is thread-local and panic-safe; ordinary +/// manifest overrides remain side-effect free. +pub(crate) fn with_install_prefetch( + cfg: &Config, + operation: impl FnOnce() -> Result, +) -> Result { + INSTALL_PREFETCH_CONFIG.with(|slot| { + let mut slot = slot.borrow_mut(); + if slot.is_some() { + return Err(anyhow!( + "an install prefetch context is already active on this thread" + )); + } + *slot = Some(cfg.clone()); + Ok(()) + })?; + let guard = InstallPrefetchGuard; + let result = operation(); + drop(guard); + result +} + /// Run one installer operation with an in-memory root manifest. Package /// manifests read from the store still come from disk; only the exact consumer /// directory is overridden. The guard is thread-local and panic-safe, so a @@ -189,10 +224,17 @@ pub(crate) fn with_manifest_override( "a manifest override is already active on this thread" )); } - *slot = Some(ManifestOverride { project, text }); + *slot = Some(ManifestOverride { + project: project.clone(), + text, + }); Ok(()) })?; let guard = ManifestOverrideGuard; + let prefetch_cfg = INSTALL_PREFETCH_CONFIG.with(|slot| slot.borrow().clone()); + if let Some(cfg) = prefetch_cfg { + crate::install_graph::prefetch(&project, &cfg, false)?; + } let result = operation(); drop(guard); result @@ -282,6 +324,32 @@ url = "https://localhost/manifestless/consumer" assert!(error.to_string().contains("already active")); } + #[test] + fn install_prefetch_context_is_scoped_and_nested_contexts_fail_closed() { + let home = tempfile::tempdir().unwrap(); + let cfg = Config { + registry: "file:///unused".to_string(), + home: home.path().to_path_buf(), + token: None, + auth_url: "https://localhost/unused".to_string(), + supabase_url: None, + supabase_key: None, + interactive: false, + }; + + with_install_prefetch(&cfg, || { + let active = INSTALL_PREFETCH_CONFIG.with(|slot| slot.borrow().is_some()); + assert!(active); + let nested = with_install_prefetch(&cfg, || Ok(())).unwrap_err(); + assert!(nested.to_string().contains("already active")); + Ok(()) + }) + .unwrap(); + + let active = INSTALL_PREFETCH_CONFIG.with(|slot| slot.borrow().is_some()); + assert!(!active); + } + #[test] fn relative_home_is_resolved_from_the_invocation_directory() { let globals = Globals { diff --git a/src/install_graph.rs b/src/install_graph.rs new file mode 100644 index 00000000..2e5318b9 --- /dev/null +++ b/src/install_graph.rs @@ -0,0 +1,301 @@ +//! Recursive, bounded-concurrency artifact prefetch for `zed install`. +//! +//! The normal installer remains the authority for resolution, lockfile writes, +//! project transactions, adapters, and symlink/copy materialization. This module +//! runs immediately before it: it walks package manifests recursively, installs +//! artifacts into the content-addressed store with a five-worker queue, and lets +//! the normal install phase reuse the warm store. +//! +//! Cross-process coordination is deliberately per artifact. Waiters block in +//! the operating system's file-lock primitive instead of polling, so unrelated +//! dependencies and unrelated projects can keep making progress. + +use std::collections::{BTreeMap, BTreeSet, VecDeque}; +use std::fs; +use std::path::{Path, PathBuf}; +use std::sync::{Arc, Condvar, Mutex, MutexGuard, mpsc}; +use std::thread::{self, JoinHandle}; + +use anyhow::{Context, Result, bail}; +use fs2::FileExt; +use globset::Glob; +use zed_interfaces::lockfile::Lockfile; +use zed_interfaces::manifest::{Manifest, is_slug}; +use zed_interfaces::paths::{LOCKFILE_FILE, MANIFEST_FILE}; +use zed_interfaces::registry::VersionMetadata; +use zed_interfaces::version::{self, Requirement}; + +use crate::config::{Config, read_manifest}; +use crate::pack::sha256_file; +use crate::registry::{Registry, registry_for}; +use crate::store::{Store, require_sha256}; + +/// Default number of artifact installs allowed to make progress at once. +pub const DEFAULT_INSTALL_CONCURRENCY: usize = 5; +const MAX_INSTALL_CONCURRENCY: usize = 32; + +#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)] +pub struct PrefetchReport { + /// Unique registry packages resolved (workspace links are not counted). + pub resolved: usize, + /// Artifacts that were absent locally and downloaded by this process. + pub downloaded: usize, +} + +#[derive(Debug, Clone)] +struct DependencyRequest { + key: String, + requirement: String, +} + +#[derive(Debug, Clone)] +struct FetchTask { + sequence: usize, + key: String, + version: VersionMetadata, +} + +#[derive(Debug)] +struct FetchResult { + dependencies: BTreeMap, + downloaded: bool, +} + +#[derive(Debug)] +struct FetchMessage { + sequence: usize, + result: Result, +} + +#[derive(Debug, Default)] +struct QueueState { + tasks: VecDeque, + closed: bool, +} + +#[derive(Debug, Default)] +struct TaskQueue { + state: Mutex, + ready: Condvar, +} + +impl TaskQueue { + fn state(&self) -> MutexGuard<'_, QueueState> { + self.state + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + } + + fn push(&self, task: FetchTask) -> bool { + let mut state = self.state(); + if state.closed { + return false; + } + state.tasks.push_back(task); + self.ready.notify_one(); + true + } + + fn pop(&self) -> Option { + let mut state = self.state(); + loop { + if let Some(task) = state.tasks.pop_front() { + return Some(task); + } + if state.closed { + return None; + } + state = self + .ready + .wait(state) + .unwrap_or_else(|poisoned| poisoned.into_inner()); + } + } + + fn close(&self) { + let mut state = self.state(); + state.closed = true; + self.ready.notify_all(); + } + + fn cancel(&self) { + let mut state = self.state(); + state.tasks.clear(); + state.closed = true; + self.ready.notify_all(); + } +} + +struct FetchPool { + queue: Arc, + results: mpsc::Receiver, + workers: Vec>, +} + +impl FetchPool { + fn new(concurrency: usize, registry_url: &str, home: &Path) -> Result { + let queue = Arc::new(TaskQueue::default()); + let (result_tx, results) = mpsc::channel(); + let mut workers = Vec::with_capacity(concurrency); + + for index in 0..concurrency { + let worker_queue = Arc::clone(&queue); + let worker_tx = result_tx.clone(); + let worker_registry = registry_url.to_string(); + let worker_home = home.to_path_buf(); + let spawn = thread::Builder::new() + .name(format!("zed-install-{index}")) + .spawn(move || worker_loop(worker_queue, worker_tx, worker_registry, worker_home)); + match spawn { + Ok(worker) => workers.push(worker), + Err(error) => { + queue.cancel(); + for worker in workers { + let _ = worker.join(); + } + return Err(error).context("starting recursive install worker"); + } + } + } + drop(result_tx); + Ok(Self { + queue, + results, + workers, + }) + } + + fn submit(&self, task: FetchTask) -> Result<()> { + if !self.queue.push(task) { + bail!("recursive install worker queue closed unexpectedly"); + } + Ok(()) + } + + fn receive(&self) -> Result { + self.results + .recv() + .context("all recursive install workers stopped unexpectedly") + } + + fn shutdown(self, cancel: bool) -> Result<()> { + if cancel { + self.queue.cancel(); + } else { + self.queue.close(); + } + let mut panicked = false; + for worker in self.workers { + panicked |= worker.join().is_err(); + } + if panicked { + bail!("a recursive install worker panicked"); + } + Ok(()) + } +} + +/// A process lock for one content hash. `lock_exclusive` blocks in the kernel; +/// it does not wake up periodically to retry. The file descriptor owns the +/// lock, so process exit or panic releases it automatically. +struct ArtifactProcessLock { + file: fs::File, +} + +impl ArtifactProcessLock { + fn acquire(home: &Path, sha256: &str) -> Result { + require_sha256(sha256)?; + let path = home.join("locks").join(format!("artifact-{sha256}.lock")); + if let Some(parent) = path.parent() { + fs::create_dir_all(parent)?; + } + let file = fs::OpenOptions::new() + .create(true) + .read(true) + .write(true) + .truncate(false) + .open(&path) + .with_context(|| format!("opening artifact lock {}", path.display()))?; + FileExt::lock_exclusive(&file) + .with_context(|| format!("waiting for artifact lock {}", path.display()))?; + Ok(Self { file }) + } +} + +impl Drop for ArtifactProcessLock { + fn drop(&mut self) { + let _ = FileExt::unlock(&self.file); + } +} + +#[derive(Debug, Default)] +struct WorkspaceGraph { + dependencies: BTreeMap>, +} + +impl WorkspaceGraph { + fn discover(project: &Path) -> Self { + let mut current = Some(project); + while let Some(directory) = current { + if directory.join(MANIFEST_FILE).is_file() + && let Ok(manifest) = read_manifest(directory) + && let Some(workspace) = manifest.workspace.as_ref() + { + return Self::collect(directory, &workspace.members); + } + current = directory.parent(); + } + Self::default() + } + + fn collect(root: &Path, patterns: &[String]) -> Self { + let mut graph = Self::default(); + for pattern in patterns { + let mut candidates = vec![root.to_path_buf()]; + for segment in pattern.split('/') { + let mut next = Vec::new(); + for base in &candidates { + if segment.contains('*') { + let Ok(glob) = Glob::new(segment) else { + continue; + }; + let matcher = glob.compile_matcher(); + if let Ok(entries) = fs::read_dir(base) { + for entry in entries.flatten() { + let name = entry.file_name(); + if entry.path().is_dir() + && matcher.is_match(Path::new(&name)) + && !name.to_string_lossy().starts_with('.') + { + next.push(entry.path()); + } + } + } + } else { + let candidate = base.join(segment); + if candidate.is_dir() { + next.push(candidate); + } + } + } + candidates = next; + } + for member_dir in candidates { + if let Ok(member) = read_manifest(&member_dir) { + let key = member.full_name(); + graph.dependencies.insert(key, member.dependencies); + } + } + } + graph + } +} + +mod artifact; +mod resolver; +#[cfg(test)] +mod tests; + +pub(crate) use artifact::ensure_artifact; +use artifact::worker_loop; +pub use resolver::prefetch; diff --git a/src/install_graph/artifact.rs b/src/install_graph/artifact.rs new file mode 100644 index 00000000..1912e673 --- /dev/null +++ b/src/install_graph/artifact.rs @@ -0,0 +1,230 @@ +use super::*; + +pub(super) fn worker_loop( + queue: Arc, + results: mpsc::Sender, + registry_url: String, + home: PathBuf, +) { + let store = Store::new(&home); + let mut registry: Option> = None; + + while let Some(task) = queue.pop() { + let sequence = task.sequence; + let key = task.key.clone(); + let message = run_fetch_task(sequence, &key, || -> Result { + if registry.is_none() { + registry = Some(registry_for(®istry_url)?); + } + let registry = registry + .as_deref() + .context("recursive install worker has no registry")?; + prefetch_one(registry, &store, task) + }); + if results.send(message).is_err() { + return; + } + } +} + +/// Convert a task panic into the same sequenced result channel used for normal +/// failures. Without this boundary, the worker would unwind after popping the +/// task, permanently losing that sequence while the coordinator waited for a +/// result that could never arrive. +pub(super) fn run_fetch_task(sequence: usize, key: &str, work: F) -> FetchMessage +where + F: FnOnce() -> Result, +{ + let result = + std::panic::catch_unwind(std::panic::AssertUnwindSafe(work)).unwrap_or_else(|payload| { + let detail = payload + .downcast_ref::<&str>() + .copied() + .or_else(|| payload.downcast_ref::().map(String::as_str)) + .unwrap_or("non-string panic payload"); + Err(anyhow::anyhow!( + "recursive install worker panicked while processing {} (sequence {}): {}", + key, + sequence, + detail + )) + }); + FetchMessage { sequence, result } +} + +fn prefetch_one(registry: &dyn Registry, store: &Store, task: FetchTask) -> Result { + let (package_dir, downloaded) = ensure_artifact(registry, store, &task.version) + .with_context(|| format!("prefetching {}@{}", task.key, task.version.version))?; + let dependencies = read_manifest(&package_dir) + .map(|manifest| manifest.dependencies) + .unwrap_or_default(); + Ok(FetchResult { + dependencies, + downloaded, + }) +} + +/// Acquire one immutable artifact through the shared cache/store path. +/// +/// Every caller uses the same blocking per-hash process lock, staged download, +/// integrity check, atomic cache publication, and extraction sequence. This is +/// intentionally crate-visible so the legacy transactional installer cannot +/// bypass recursive-prefetch locking. +pub(crate) fn ensure_artifact( + registry: &dyn Registry, + store: &Store, + version: &VersionMetadata, +) -> Result<(PathBuf, bool)> { + validate_version_identity(version, &version.org, &version.name, &version.version)?; + if store.has(&version.sha256) { + return Ok((store.pkg_dir(&version.sha256), false)); + } + + let _artifact_lock = ArtifactProcessLock::acquire(store.home(), &version.sha256)?; + if store.has(&version.sha256) { + return Ok((store.pkg_dir(&version.sha256), false)); + } + + let cached = store.cached_artifact(&version.sha256); + let mut downloaded = false; + if !cached.is_file() { + download_atomic(registry, version, &cached)?; + downloaded = true; + } + + match store.add_artifact(&cached, &version.sha256) { + Ok(package_dir) => Ok((package_dir, downloaded)), + Err(first_error) if !downloaded => { + // A killed legacy client may have left a partial cache file. The + // per-artifact lock makes removal and replacement safe. + let _ = fs::remove_file(&cached); + download_atomic(registry, version, &cached)?; + store + .add_artifact(&cached, &version.sha256) + .with_context(|| { + format!("cached artifact was invalid ({first_error:#}); redownload also failed") + }) + .map(|package_dir| (package_dir, true)) + } + Err(error) => Err(error), + } +} + +fn download_atomic( + registry: &dyn Registry, + version: &VersionMetadata, + cached: &Path, +) -> Result<()> { + let cache_dir = cached.parent().context("cached artifact has a parent")?; + fs::create_dir_all(cache_dir)?; + let staging_dir = tempfile::tempdir_in(cache_dir)?; + let staged = staging_dir.path().join("artifact.download"); + registry.download(version, &staged)?; + + let (actual_sha256, _) = sha256_file(&staged)?; + if actual_sha256 != version.sha256 { + bail!( + "artifact hash mismatch: expected {}, got {}", + version.sha256, + actual_sha256 + ); + } + + if cached.exists() { + let (cached_sha256, _) = sha256_file(cached)?; + if cached_sha256 == version.sha256 { + return Ok(()); + } + fs::remove_file(cached)?; + } + match fs::rename(&staged, cached) { + Ok(()) => Ok(()), + Err(error) if cached.is_file() => { + let (cached_sha256, _) = sha256_file(cached)?; + if cached_sha256 == version.sha256 { + Ok(()) + } else { + Err(error).with_context(|| { + format!("publishing downloaded artifact to {}", cached.display()) + }) + } + } + Err(error) => Err(error) + .with_context(|| format!("publishing downloaded artifact to {}", cached.display())), + } +} + +pub(super) fn split_key(key: &str) -> Result<(String, String)> { + let mut parts = key.splitn(2, '/'); + match (parts.next(), parts.next()) { + (Some(org), Some(name)) if !org.is_empty() && !name.is_empty() => { + Ok((org.to_string(), name.to_string())) + } + _ => bail!("invalid package spec `{key}` (expected org/name)"), + } +} + +pub(super) fn validate_version_identity( + version: &VersionMetadata, + expected_org: &str, + expected_name: &str, + expected_version: &str, +) -> Result<()> { + if !is_slug(&version.org) || !is_slug(&version.name) { + bail!( + "registry returned invalid package identity `{}/{}`; refusing", + version.org, + version.name + ); + } + require_sha256(&version.sha256)?; + if version.org != expected_org + || version.name != expected_name + || version.version != expected_version + { + bail!( + "registry returned `{}/{}@{}` while resolving `{expected_org}/{expected_name}@{expected_version}`; refusing", + version.org, + version.name, + version.version + ); + } + Ok(()) +} + +pub(super) fn resolve_version( + registry: &dyn Registry, + org: &str, + name: &str, + requirement_text: &str, +) -> Result { + let key = format!("{org}/{name}"); + let requirement = Requirement::parse(requirement_text); + let package = registry.get_package(org, name)?; + let mut candidates = package.versions.clone(); + let mut skipped_yanked = Vec::new(); + + loop { + let selected = version::resolve(&requirement, &candidates).map(str::to_string); + let Some(selected) = selected else { + if !skipped_yanked.is_empty() { + bail!( + "the only version(s) of {key} satisfying `{requirement_text}` are yanked ({}); existing lockfiles keep working via `zed install --frozen`", + skipped_yanked.join(", ") + ); + } + bail!( + "no version of {key} satisfies `{requirement_text}` (available: {})", + package.versions.join(", ") + ); + }; + let version = registry.get_version(org, name, &selected)?; + validate_version_identity(&version, org, name, &selected)?; + if version.yanked { + candidates.retain(|candidate| *candidate != selected); + skipped_yanked.push(selected); + continue; + } + return Ok(version); + } +} diff --git a/src/install_graph/resolver.rs b/src/install_graph/resolver.rs new file mode 100644 index 00000000..7fc169e4 --- /dev/null +++ b/src/install_graph/resolver.rs @@ -0,0 +1,231 @@ +use super::artifact::{resolve_version, split_key, validate_version_identity}; +use super::*; + +/// Warm the content-addressed store for an install. The normal installer runs +/// immediately afterward and remains responsible for the project transaction. +pub fn prefetch(project: &Path, cfg: &Config, frozen: bool) -> Result { + let concurrency = install_concurrency(); + let registry = registry_for(&cfg.registry)?; + let report = if frozen { + prefetch_locked(project, cfg, registry.as_ref(), concurrency)? + } else { + let manifest = read_manifest(project)?; + prefetch_recursive(project, cfg, registry.as_ref(), &manifest, concurrency)? + }; + + if report.resolved > 0 { + eprintln!( + "recursive install prefetch: {} package(s), up to {} concurrent, {} downloaded", + report.resolved, concurrency, report.downloaded + ); + } + Ok(report) +} + +fn install_concurrency() -> usize { + normalize_concurrency(std::env::var("ZED_PKG_INSTALL_CONCURRENCY").ok().as_deref()) +} + +pub(super) fn normalize_concurrency(raw: Option<&str>) -> usize { + raw.and_then(|value| value.parse::().ok()) + .filter(|value| *value > 0) + .unwrap_or(DEFAULT_INSTALL_CONCURRENCY) + .min(MAX_INSTALL_CONCURRENCY) +} + +fn run_with_pool( + cfg: &Config, + concurrency: usize, + operation: impl FnOnce(&FetchPool) -> Result, +) -> Result { + let pool = FetchPool::new(concurrency, &cfg.registry, &cfg.home)?; + match operation(&pool) { + Ok(value) => { + pool.shutdown(false)?; + Ok(value) + } + Err(error) => { + let _ = pool.shutdown(true); + Err(error) + } + } +} + +fn prefetch_locked( + project: &Path, + cfg: &Config, + registry: &dyn Registry, + concurrency: usize, +) -> Result { + let lock_path = project.join(LOCKFILE_FILE); + let text = fs::read_to_string(&lock_path) + .with_context(|| format!("--frozen requires {}", lock_path.display()))?; + let lock = Lockfile::parse(&text)?; + let mut tasks = Vec::with_capacity(lock.packages.len()); + let mut seen = BTreeSet::new(); + + for locked in &lock.packages { + if !is_slug(&locked.org) || !is_slug(&locked.name) { + bail!( + "lockfile entry `{}/{}` has an invalid identity; refusing", + locked.org, + locked.name + ); + } + require_sha256(&locked.sha256)?; + let key = locked.full_name(); + if !seen.insert(key.clone()) { + bail!("duplicate package `{key}` in {LOCKFILE_FILE}"); + } + let version = registry.get_version(&locked.org, &locked.name, &locked.version)?; + validate_version_identity(&version, &locked.org, &locked.name, &locked.version)?; + if version.sha256 != locked.sha256 { + bail!( + "registry artifact for {}@{} changed (lock {} vs registry {}); refusing", + key, + locked.version, + locked.sha256, + version.sha256 + ); + } + tasks.push(FetchTask { + sequence: tasks.len(), + key, + version, + }); + } + + if tasks.is_empty() { + return Ok(PrefetchReport::default()); + } + + run_with_pool(cfg, concurrency, |pool| { + let total = tasks.len(); + for task in tasks { + pool.submit(task)?; + } + let mut downloaded = 0usize; + for _ in 0..total { + let message = pool.receive()?; + let result = message.result?; + downloaded += usize::from(result.downloaded); + } + Ok(PrefetchReport { + resolved: total, + downloaded, + }) + }) +} + +fn prefetch_recursive( + project: &Path, + cfg: &Config, + registry: &dyn Registry, + manifest: &Manifest, + concurrency: usize, +) -> Result { + if manifest.dependencies.is_empty() { + return Ok(PrefetchReport::default()); + } + let workspace = WorkspaceGraph::discover(project); + + run_with_pool(cfg, concurrency, |pool| { + let mut pending: VecDeque = manifest + .dependencies + .iter() + .map(|(key, requirement)| DependencyRequest { + key: key.clone(), + requirement: requirement.clone(), + }) + .collect(); + let mut resolved: BTreeMap = BTreeMap::new(); + let mut expanded_workspace = BTreeSet::new(); + let mut in_flight = 0usize; + let mut next_sequence = 0usize; + let mut next_result = 0usize; + let mut buffered_results: BTreeMap> = BTreeMap::new(); + let mut downloaded = 0usize; + + loop { + while let Some(request) = pending.pop_front() { + if let Some(member_dependencies) = workspace.dependencies.get(&request.key) { + if expanded_workspace.insert(request.key.clone()) { + pending.extend(member_dependencies.iter().map(|(key, requirement)| { + DependencyRequest { + key: key.clone(), + requirement: requirement.clone(), + } + })); + } + continue; + } + + let (org, name) = split_key(&request.key)?; + let requirement = Requirement::parse(&request.requirement); + if let Some(existing) = resolved.get(&request.key) { + if requirement.matches(&existing.version) { + continue; + } + bail!( + "version conflict for {}: resolved {} but another dependency requires `{}` (zed installs one version per package)", + request.key, + existing.version, + request.requirement + ); + } + + let version = resolve_version(registry, &org, &name, &request.requirement)?; + resolved.insert(request.key.clone(), version.clone()); + pool.submit(FetchTask { + sequence: next_sequence, + key: request.key, + version, + })?; + next_sequence += 1; + in_flight += 1; + } + + if in_flight == 0 { + break; + } + + let result = receive_in_order(pool, &mut buffered_results, next_result)?; + next_result += 1; + in_flight -= 1; + downloaded += usize::from(result.downloaded); + pending.extend( + result + .dependencies + .into_iter() + .map(|(key, requirement)| DependencyRequest { key, requirement }), + ); + } + + Ok(PrefetchReport { + resolved: resolved.len(), + downloaded, + }) + }) +} + +fn receive_in_order( + pool: &FetchPool, + buffered: &mut BTreeMap>, + expected: usize, +) -> Result { + loop { + if let Some(result) = buffered.remove(&expected) { + return result; + } + let message = pool.receive()?; + if message.sequence == expected { + return message.result; + } + if buffered.insert(message.sequence, message.result).is_some() { + bail!( + "recursive install worker returned duplicate result sequence {}", + message.sequence + ); + } + } +} diff --git a/src/install_graph/tests.rs b/src/install_graph/tests.rs new file mode 100644 index 00000000..d07a62d3 --- /dev/null +++ b/src/install_graph/tests.rs @@ -0,0 +1,653 @@ +use std::sync::{ + Arc, Barrier, + atomic::{AtomicUsize, Ordering}, + mpsc as test_mpsc, +}; +use std::thread; +use std::time::Duration; + +use zed_interfaces::lockfile::LockedPackage; +use zed_interfaces::registry::{PackageMetadata, VersionMetadata}; +use zed_interfaces::vcs::Vcs; + +use super::resolver::normalize_concurrency; +use super::*; +use crate::pack::pack; + +fn test_config(registry: &Path, home: &Path) -> Config { + Config { + registry: format!("file://{}", registry.display()), + home: home.to_path_buf(), + token: None, + auth_url: "http://127.0.0.1/unused".to_string(), + supabase_url: None, + supabase_key: None, + interactive: false, + } +} + +fn manifest_text(org: &str, name: &str, version: &str, dependencies: &[(&str, &str)]) -> String { + let mut text = format!( + r#"[package] +org = "{org}" +name = "{name}" +version = "{version}" + +[package.repository] +vcs = "git" +url = "https://example.invalid/{org}/{name}" +"#, + ); + if !dependencies.is_empty() { + text.push_str("\n[dependencies]\n"); + for (key, requirement) in dependencies { + text.push_str(&format!("\"{key}\" = \"{requirement}\"\n")); + } + } + text +} + +fn publish_fixture( + registry_root: &Path, + scratch: &Path, + org: &str, + name: &str, + version: &str, + dependencies: &[(&str, &str)], +) -> String { + let source = scratch.join(format!("source-{name}")); + fs::create_dir_all(&source).unwrap(); + let manifest_text = manifest_text(org, name, version, dependencies); + fs::write(source.join(MANIFEST_FILE), &manifest_text).unwrap(); + fs::write( + source.join("payload.txt"), + format!("{org}/{name}@{version}\n"), + ) + .unwrap(); + let manifest = Manifest::parse(&manifest_text).unwrap(); + let packed = pack( + &source, + &manifest, + Some(&scratch.join(format!("packed-{name}"))), + ) + .unwrap(); + + let artifacts = registry_root.join("artifacts"); + fs::create_dir_all(&artifacts).unwrap(); + fs::copy( + &packed.path, + artifacts.join(format!("{}.tar.gz", packed.sha256)), + ) + .unwrap(); + + let package_dir = registry_root.join("packages").join(org).join(name); + fs::create_dir_all(package_dir.join("versions")).unwrap(); + let version_metadata = VersionMetadata { + org: org.to_string(), + name: name.to_string(), + version: version.to_string(), + sha256: packed.sha256.clone(), + size: packed.size, + format: packed.format, + vcs_tag: format!("v{version}"), + vcs_commit: Some("0123456789abcdef0123456789abcdef01234567".to_string()), + download_url: format!("file://{}/{}.tar.gz", artifacts.display(), packed.sha256), + published_at: "1970-01-01T00:00:00Z".to_string(), + yanked: false, + }; + fs::write( + package_dir.join("versions").join(format!("{version}.json")), + serde_json::to_string_pretty(&version_metadata).unwrap(), + ) + .unwrap(); + let package_metadata = PackageMetadata { + org: org.to_string(), + name: name.to_string(), + description: Some(format!("fixture {name}")), + vcs: Vcs::Git, + repo_url: format!("https://example.invalid/{org}/{name}"), + version_scheme: manifest.package.version_scheme, + latest: Some(version.to_string()), + tags: Vec::new(), + versions: vec![version.to_string()], + }; + fs::write( + package_dir.join("package.json"), + serde_json::to_string_pretty(&package_metadata).unwrap(), + ) + .unwrap(); + packed.sha256 +} + +fn published_version( + registry_root: &Path, + org: &str, + name: &str, + version: &str, +) -> VersionMetadata { + let path = registry_root + .join("packages") + .join(org) + .join(name) + .join("versions") + .join(format!("{version}.json")); + serde_json::from_str(&fs::read_to_string(path).unwrap()).unwrap() +} + +struct CountingRegistry { + artifact: PathBuf, + downloads: Arc, + started: Option>, + release: Option>, + fail_after_write: bool, +} + +impl Registry for CountingRegistry { + fn get_package(&self, _org: &str, _name: &str) -> Result { + unreachable!("mixed-path acquisition test does not resolve package metadata") + } + + fn get_version(&self, _org: &str, _name: &str, _version: &str) -> Result { + unreachable!("mixed-path acquisition test receives resolved version metadata") + } + + fn download(&self, _version: &VersionMetadata, dest: &Path) -> Result<()> { + self.downloads.fetch_add(1, Ordering::SeqCst); + if let Some(started) = &self.started { + started.send(()).unwrap(); + } + if let Some(release) = &self.release { + release.recv_timeout(Duration::from_secs(5)).unwrap(); + } + fs::create_dir_all(dest.parent().unwrap())?; + if self.fail_after_write { + fs::write(dest, b"partial fixture download")?; + bail!("injected artifact download failure"); + } + fs::copy(&self.artifact, dest)?; + Ok(()) + } + + fn publish( + &self, + _meta: &zed_interfaces::registry::PublishMeta, + _artifact: &Path, + _token: Option<&str>, + ) -> Result { + unreachable!("mixed-path acquisition test does not publish") + } + + fn claim_org( + &self, + _slug: &str, + _token: Option<&str>, + ) -> Result { + unreachable!("mixed-path acquisition test does not claim organizations") + } + + fn search(&self, _query: &str) -> Result { + unreachable!("mixed-path acquisition test does not search") + } + + fn yank( + &self, + _org: &str, + _name: &str, + _version: &str, + _yanked: bool, + _token: Option<&str>, + ) -> Result { + unreachable!("mixed-path acquisition test does not yank") + } + + fn audit_log( + &self, + _org: &str, + _limit: Option, + _token: Option<&str>, + ) -> Result { + unreachable!("mixed-path acquisition test does not read audit logs") + } +} + +#[test] +fn legacy_and_recursive_acquisition_paths_share_one_download_lock() { + let temp = tempfile::tempdir().unwrap(); + let registry_root = temp.path().join("registry"); + let scratch = temp.path().join("scratch"); + let home = temp.path().join("home"); + + let sha = publish_fixture(®istry_root, &scratch, "test", "shared", "1.0.0", &[]); + let version = published_version(®istry_root, "test", "shared", "1.0.0"); + let artifact = registry_root + .join("artifacts") + .join(format!("{sha}.tar.gz")); + let downloads = Arc::new(AtomicUsize::new(0)); + let (recursive_started_tx, recursive_started_rx) = test_mpsc::channel(); + let (recursive_release_tx, recursive_release_rx) = test_mpsc::channel(); + let (legacy_started_tx, legacy_started_rx) = test_mpsc::channel(); + + let recursive_registry = CountingRegistry { + artifact: artifact.clone(), + downloads: Arc::clone(&downloads), + started: Some(recursive_started_tx), + release: Some(recursive_release_rx), + fail_after_write: false, + }; + let legacy_registry = CountingRegistry { + artifact, + downloads: Arc::clone(&downloads), + started: Some(legacy_started_tx), + release: None, + fail_after_write: false, + }; + let recursive_store = Store::new(&home); + let legacy_store = Store::new(&home); + let recursive_version = version.clone(); + let legacy_version = version; + + let recursive = thread::spawn(move || { + super::ensure_artifact(&recursive_registry, &recursive_store, &recursive_version).unwrap() + }); + recursive_started_rx + .recv_timeout(Duration::from_secs(2)) + .unwrap(); + + let legacy = thread::spawn(move || { + crate::ops::legacy_ensure_artifact_for_test( + &legacy_registry, + &legacy_store, + &legacy_version, + ) + .unwrap() + }); + + let legacy_entered_download = legacy_started_rx + .recv_timeout(Duration::from_millis(300)) + .is_ok(); + recursive_release_tx.send(()).unwrap(); + + let (recursive_path, recursive_downloaded) = recursive.join().unwrap(); + let legacy_path = legacy.join().unwrap(); + assert!(recursive_downloaded); + assert!( + !legacy_entered_download, + "the legacy path bypassed the shared artifact lock and started a duplicate download" + ); + assert_eq!(downloads.load(Ordering::SeqCst), 1); + assert_eq!(legacy_path, recursive_path); + assert!(recursive_path.is_dir()); +} + +#[test] +fn worker_panics_are_reported_without_losing_the_task_sequence() { + let message = super::artifact::run_fetch_task(17, "test/panic", || -> Result { + panic!("fixture worker panic") + }); + + assert_eq!(message.sequence, 17); + let error = message.result.unwrap_err().to_string(); + assert!(error.contains("test/panic"), "{error}"); + assert!(error.contains("sequence 17"), "{error}"); + assert!(error.contains("fixture worker panic"), "{error}"); +} + +#[test] +fn failed_downloads_remove_staging_files_and_never_publish_cache_entries() { + let temp = tempfile::tempdir().unwrap(); + let registry_root = temp.path().join("registry"); + let scratch = temp.path().join("scratch"); + let home = temp.path().join("home"); + + let sha = publish_fixture(®istry_root, &scratch, "test", "failure", "1.0.0", &[]); + let version = published_version(®istry_root, "test", "failure", "1.0.0"); + let registry = CountingRegistry { + artifact: registry_root + .join("artifacts") + .join(format!("{sha}.tar.gz")), + downloads: Arc::new(AtomicUsize::new(0)), + started: None, + release: None, + fail_after_write: true, + }; + let store = Store::new(&home); + + let error = super::ensure_artifact(®istry, &store, &version) + .unwrap_err() + .to_string(); + assert!( + error.contains("injected artifact download failure"), + "{error}" + ); + assert!(!store.cached_artifact(&sha).exists()); + assert!(!store.has(&sha)); + let cache_entries = fs::read_dir(store.cache_dir()) + .map(|entries| entries.count()) + .unwrap_or(0); + assert_eq!(cache_entries, 0, "failed staging directory leaked"); +} + +#[test] +fn defaults_to_five_workers_and_bounds_overrides() { + assert_eq!(normalize_concurrency(None), 5); + assert_eq!(normalize_concurrency(Some("not-a-number")), 5); + assert_eq!(normalize_concurrency(Some("0")), 5); + assert_eq!(normalize_concurrency(Some("1")), 1); + assert_eq!(normalize_concurrency(Some("999")), MAX_INSTALL_CONCURRENCY); +} + +#[test] +fn recursively_prefetches_packages_of_packages_and_deduplicates_diamonds() { + let temp = tempfile::tempdir().unwrap(); + let registry = temp.path().join("registry"); + let scratch = temp.path().join("scratch"); + let home = temp.path().join("home"); + let project = temp.path().join("project"); + fs::create_dir_all(&project).unwrap(); + let leaf = publish_fixture(®istry, &scratch, "test", "leaf", "1.0.0", &[]); + let left = publish_fixture( + ®istry, + &scratch, + "test", + "left", + "1.0.0", + &[("test/leaf", "^1")], + ); + let right = publish_fixture( + ®istry, + &scratch, + "test", + "right", + "1.0.0", + &[("test/leaf", "^1")], + ); + let root = publish_fixture( + ®istry, + &scratch, + "test", + "root", + "1.0.0", + &[("test/left", "^1"), ("test/right", "^1")], + ); + fs::write( + project.join(MANIFEST_FILE), + manifest_text("consumer", "app", "0.1.0", &[("test/root", "^1")]), + ) + .unwrap(); + + let report = prefetch(&project, &test_config(®istry, &home), false).unwrap(); + assert_eq!(report.resolved, 4); + assert_eq!(report.downloaded, 4); + let store = Store::new(&home); + for sha in [leaf, left, right, root] { + assert!(store.has(&sha), "missing prefetched artifact {sha}"); + } +} + +#[test] +fn dependency_cycles_terminate_after_each_package_is_resolved_once() { + let temp = tempfile::tempdir().unwrap(); + let registry = temp.path().join("registry"); + let scratch = temp.path().join("scratch"); + let home = temp.path().join("home"); + let project = temp.path().join("project"); + fs::create_dir_all(&project).unwrap(); + + publish_fixture( + ®istry, + &scratch, + "test", + "a", + "1.0.0", + &[("test/b", "^1")], + ); + publish_fixture( + ®istry, + &scratch, + "test", + "b", + "1.0.0", + &[("test/a", "^1")], + ); + fs::write( + project.join(MANIFEST_FILE), + manifest_text("consumer", "app", "0.1.0", &[("test/a", "^1")]), + ) + .unwrap(); + + let report = prefetch(&project, &test_config(®istry, &home), false).unwrap(); + assert_eq!(report.resolved, 2); + assert_eq!(report.downloaded, 2); +} + +#[test] +fn concurrent_prefetches_share_one_artifact_download() { + let temp = tempfile::tempdir().unwrap(); + let registry = temp.path().join("registry"); + let scratch = temp.path().join("scratch"); + let home = temp.path().join("home"); + let first_project = temp.path().join("first-project"); + let second_project = temp.path().join("second-project"); + fs::create_dir_all(&first_project).unwrap(); + fs::create_dir_all(&second_project).unwrap(); + + publish_fixture(®istry, &scratch, "test", "leaf", "1.0.0", &[]); + let consumer_manifest = manifest_text("consumer", "app", "0.1.0", &[("test/leaf", "^1")]); + fs::write(first_project.join(MANIFEST_FILE), &consumer_manifest).unwrap(); + fs::write(second_project.join(MANIFEST_FILE), &consumer_manifest).unwrap(); + + let barrier = Arc::new(Barrier::new(3)); + let first_cfg = test_config(®istry, &home); + let second_cfg = first_cfg.clone(); + let first_barrier = Arc::clone(&barrier); + let second_barrier = Arc::clone(&barrier); + let first = thread::spawn(move || { + first_barrier.wait(); + prefetch(&first_project, &first_cfg, false).unwrap() + }); + let second = thread::spawn(move || { + second_barrier.wait(); + prefetch(&second_project, &second_cfg, false).unwrap() + }); + + barrier.wait(); + let first = first.join().unwrap(); + let second = second.join().unwrap(); + assert_eq!(first.resolved + second.resolved, 2); + assert_eq!( + first.downloaded + second.downloaded, + 1, + "one process should publish the shared content hash and the other should reuse it" + ); +} + +#[test] +fn a_warm_store_resolves_the_graph_without_redownloading_artifacts() { + let temp = tempfile::tempdir().unwrap(); + let registry = temp.path().join("registry"); + let scratch = temp.path().join("scratch"); + let home = temp.path().join("home"); + let project = temp.path().join("project"); + fs::create_dir_all(&project).unwrap(); + + publish_fixture(®istry, &scratch, "test", "leaf", "1.0.0", &[]); + publish_fixture( + ®istry, + &scratch, + "test", + "left", + "1.0.0", + &[("test/leaf", "^1")], + ); + publish_fixture( + ®istry, + &scratch, + "test", + "right", + "1.0.0", + &[("test/leaf", "^1")], + ); + publish_fixture( + ®istry, + &scratch, + "test", + "root", + "1.0.0", + &[("test/left", "^1"), ("test/right", "^1")], + ); + fs::write( + project.join(MANIFEST_FILE), + manifest_text("consumer", "app", "0.1.0", &[("test/root", "^1")]), + ) + .unwrap(); + let cfg = test_config(®istry, &home); + + let cold = prefetch(&project, &cfg, false).unwrap(); + let warm = prefetch(&project, &cfg, false).unwrap(); + assert_eq!(cold.resolved, 4); + assert_eq!(cold.downloaded, 4); + assert_eq!(warm.resolved, 4); + assert_eq!(warm.downloaded, 0); +} + +#[test] +fn conflicting_transitive_requirements_fail_in_deterministic_graph_order() { + let temp = tempfile::tempdir().unwrap(); + let registry = temp.path().join("registry"); + let scratch = temp.path().join("scratch"); + let home = temp.path().join("home"); + let project = temp.path().join("project"); + fs::create_dir_all(&project).unwrap(); + + publish_fixture(®istry, &scratch, "test", "leaf", "1.0.0", &[]); + publish_fixture( + ®istry, + &scratch, + "test", + "left", + "1.0.0", + &[("test/leaf", "^1")], + ); + publish_fixture( + ®istry, + &scratch, + "test", + "right", + "1.0.0", + &[("test/leaf", "^2")], + ); + publish_fixture( + ®istry, + &scratch, + "test", + "root", + "1.0.0", + &[("test/left", "^1"), ("test/right", "^1")], + ); + fs::write( + project.join(MANIFEST_FILE), + manifest_text("consumer", "app", "0.1.0", &[("test/root", "^1")]), + ) + .unwrap(); + + let error = prefetch(&project, &test_config(®istry, &home), false).unwrap_err(); + let message = format!("{error:#}"); + assert!( + message.contains("version conflict for test/leaf"), + "{message}" + ); + assert!(message.contains("resolved 1.0.0"), "{message}"); + assert!(message.contains("requires `^2`"), "{message}"); +} + +#[test] +fn frozen_prefetch_installs_every_locked_artifact_without_a_manifest() { + let temp = tempfile::tempdir().unwrap(); + let registry = temp.path().join("registry"); + let scratch = temp.path().join("scratch"); + let home = temp.path().join("home"); + let project = temp.path().join("project"); + fs::create_dir_all(&project).unwrap(); + + for (name, dependencies) in [ + ("leaf", vec![]), + ("left", vec![("test/leaf", "^1")]), + ("right", vec![("test/leaf", "^1")]), + ("root", vec![("test/left", "^1"), ("test/right", "^1")]), + ] { + publish_fixture(®istry, &scratch, "test", name, "1.0.0", &dependencies); + } + + let source = format!("file://{}", registry.display()); + let mut lock = Lockfile::default(); + for name in ["leaf", "left", "right", "root"] { + let version = published_version(®istry, "test", name, "1.0.0"); + lock.upsert(LockedPackage { + org: version.org, + name: version.name, + version: version.version, + sha256: version.sha256, + size: version.size, + format: version.format, + vcs_tag: version.vcs_tag, + vcs_commit: version.vcs_commit, + source: source.clone(), + }); + } + fs::write(project.join(LOCKFILE_FILE), lock.to_toml_string().unwrap()).unwrap(); + + let report = prefetch(&project, &test_config(®istry, &home), true).unwrap(); + assert_eq!(report.resolved, 4); + assert_eq!(report.downloaded, 4); +} + +#[test] +fn a_corrupt_partial_cache_is_replaced_under_the_artifact_lock() { + let temp = tempfile::tempdir().unwrap(); + let registry = temp.path().join("registry"); + let scratch = temp.path().join("scratch"); + let home = temp.path().join("home"); + let project = temp.path().join("project"); + fs::create_dir_all(&project).unwrap(); + + let sha = publish_fixture(®istry, &scratch, "test", "leaf", "1.0.0", &[]); + fs::write( + project.join(MANIFEST_FILE), + manifest_text("consumer", "app", "0.1.0", &[("test/leaf", "^1")]), + ) + .unwrap(); + let store = Store::new(&home); + let cached = store.cached_artifact(&sha); + fs::create_dir_all(cached.parent().unwrap()).unwrap(); + fs::write(&cached, b"partial download").unwrap(); + + let report = prefetch(&project, &test_config(®istry, &home), false).unwrap(); + assert_eq!(report.resolved, 1); + assert_eq!(report.downloaded, 1); + assert!(store.has(&sha)); + assert_eq!(sha256_file(&cached).unwrap().0, sha); +} + +#[cfg(unix)] +#[test] +fn artifact_waiters_block_until_the_owner_releases_the_kernel_lock() { + let temp = tempfile::tempdir().unwrap(); + let sha = "a".repeat(64); + let owner = ArtifactProcessLock::acquire(temp.path(), &sha).unwrap(); + let second_home = temp.path().to_path_buf(); + let second_sha = sha.clone(); + let (acquired_tx, acquired_rx) = test_mpsc::channel(); + let waiter = thread::spawn(move || { + let _waiter = ArtifactProcessLock::acquire(&second_home, &second_sha).unwrap(); + acquired_tx.send(()).unwrap(); + }); + + assert!( + acquired_rx + .recv_timeout(Duration::from_millis(100)) + .is_err() + ); + drop(owner); + acquired_rx.recv_timeout(Duration::from_secs(2)).unwrap(); + waiter.join().unwrap(); +} diff --git a/src/lib.rs b/src/lib.rs index 49076839..a4a79627 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -7,8 +7,10 @@ pub mod completion; pub mod config; pub mod dev; pub mod flags; +pub mod install_graph; pub mod interactive; pub mod manifestless; +#[path = "ops_entry.rs"] pub mod ops; pub mod pack; pub mod preflight; diff --git a/src/ops.rs b/src/ops.rs index 2c365a2c..8d4696bf 100644 --- a/src/ops.rs +++ b/src/ops.rs @@ -229,14 +229,17 @@ pub struct InstallOutcome { fn ensure_artifact(reg: &dyn Registry, store: &Store, vm: &VersionMetadata) -> Result { validate_version_metadata(vm)?; - if store.has(&vm.sha256) { - return Ok(store.pkg_dir(&vm.sha256)); - } - let cached = store.cached_artifact(&vm.sha256); - if !cached.exists() { - reg.download(vm, &cached)?; - } - store.add_artifact(&cached, &vm.sha256) + crate::install_graph::ensure_artifact(reg, store, vm) + .map(|(package_dir, _downloaded)| package_dir) +} + +#[cfg(test)] +pub(crate) fn legacy_ensure_artifact_for_test( + reg: &dyn Registry, + store: &Store, + vm: &VersionMetadata, +) -> Result { + ensure_artifact(reg, store, vm) } fn replace_dest(dest: &Path) -> Result<()> { diff --git a/src/ops_entry.rs b/src/ops_entry.rs new file mode 100644 index 00000000..36815c27 --- /dev/null +++ b/src/ops_entry.rs @@ -0,0 +1,83 @@ +//! Public installer facade. +//! +//! The existing implementation continues to own dependency resolution, +//! project transactions, lockfile writes, adapter wiring, and materialization. +//! This facade warms the content-addressed store with the recursive, +//! bounded-concurrency installer before entering that transactional phase. + +use std::path::Path; + +use anyhow::Result; + +use crate::cli::{Adapter, InstallMode}; +use crate::config::Config; + +#[path = "ops.rs"] +mod implementation; + +pub use implementation::{ + InstallOutcome, WorkspaceInfo, build_cmd, build_publish_meta, cache_clean, find, gc, init, + login, org_audit, org_claim, pack_cmd, publish, run, split_key, store_prune, store_status, + uninstall, yank, +}; + +pub(crate) use implementation::{ + detect_adapter, detect_native_manifest_target, detect_structure_target, detect_target, +}; + +#[cfg(test)] +pub(crate) use implementation::legacy_ensure_artifact_for_test; + +pub fn add(project: &Path, cfg: &Config, spec: &str) -> Result<()> { + crate::config::with_install_prefetch(cfg, || implementation::add(project, cfg, spec)) +} + +pub fn remove(project: &Path, cfg: &Config, spec: &str) -> Result<()> { + crate::config::with_install_prefetch(cfg, || implementation::remove(project, cfg, spec)) +} + +#[allow(clippy::too_many_arguments)] +pub fn install( + project: &Path, + cfg: &Config, + frozen: bool, + mode: InstallMode, + adapter: Adapter, + allow_build: bool, + target: Option<&str>, + allow_ecosystem_mismatch: bool, +) -> Result { + crate::install_graph::prefetch(project, cfg, frozen)?; + implementation::install( + project, + cfg, + frozen, + mode, + adapter, + allow_build, + target, + allow_ecosystem_mismatch, + ) +} + +#[allow(clippy::too_many_arguments)] +pub(crate) fn install_frozen_lock_only( + project: &Path, + cfg: &Config, + mode: InstallMode, + adapter: Adapter, + allow_build: bool, + target: Option<&str>, + allow_ecosystem_mismatch: bool, +) -> Result { + crate::install_graph::prefetch(project, cfg, true)?; + implementation::install_frozen_lock_only( + project, + cfg, + mode, + adapter, + allow_build, + target, + allow_ecosystem_mismatch, + ) +} diff --git a/src/store.rs b/src/store.rs index 6f4dfcce..018446af 100644 --- a/src/store.rs +++ b/src/store.rs @@ -161,6 +161,10 @@ impl Store { } } + pub(crate) fn home(&self) -> &Path { + &self.home + } + pub fn root(&self) -> PathBuf { self.home.join("store") } diff --git a/tests/recursive_install_concurrency.rs b/tests/recursive_install_concurrency.rs new file mode 100644 index 00000000..a8b2042a --- /dev/null +++ b/tests/recursive_install_concurrency.rs @@ -0,0 +1,438 @@ +use std::collections::HashMap; +use std::fs; +use std::io::{BufRead, BufReader, Write}; +use std::net::{SocketAddr, TcpListener, TcpStream}; +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; +use std::sync::{Arc, Condvar, Mutex}; +use std::thread::{self, JoinHandle}; +use std::time::{Duration, Instant}; + +use anyhow::{Context, Result}; +use zed_cli::config::Config; +use zed_cli::install_graph::prefetch; +use zed_cli::pack::pack; +use zed_cli::store::Store; +use zed_interfaces::manifest::Manifest; +use zed_interfaces::registry::{self, PackageMetadata, VersionMetadata}; +use zed_interfaces::vcs::Vcs; + +const ORG: &str = "recursive-concurrency"; +const VERSION: &str = "1.0.0"; +const PACKAGE_COUNT: usize = 12; +const EXPECTED_CONCURRENCY: usize = 5; + +#[derive(Clone)] +struct Response { + content_type: &'static str, + body: Arc>, + artifact: bool, +} + +#[derive(Default)] +struct GateState { + arrived: usize, + open: bool, + timed_out: bool, +} + +struct ServerState { + responses: HashMap, + active_artifacts: AtomicUsize, + max_active_artifacts: AtomicUsize, + artifact_requests: AtomicUsize, + handler_failed: AtomicBool, + gate: Mutex, + gate_ready: Condvar, +} + +struct RegistryServer { + address: SocketAddr, + state: Arc, + shutdown: Arc, + accept_thread: Option>, +} + +impl RegistryServer { + fn start(listener: TcpListener, responses: HashMap) -> Result { + let address = listener.local_addr()?; + let state = Arc::new(ServerState { + responses, + active_artifacts: AtomicUsize::new(0), + max_active_artifacts: AtomicUsize::new(0), + artifact_requests: AtomicUsize::new(0), + handler_failed: AtomicBool::new(false), + gate: Mutex::new(GateState::default()), + gate_ready: Condvar::new(), + }); + let shutdown = Arc::new(AtomicBool::new(false)); + let accept_state = Arc::clone(&state); + let accept_shutdown = Arc::clone(&shutdown); + let accept_thread = thread::Builder::new() + .name("recursive-registry".to_string()) + .spawn(move || { + while let Ok((stream, _)) = listener.accept() { + if accept_shutdown.load(Ordering::SeqCst) { + break; + } + let handler_state = Arc::clone(&accept_state); + thread::spawn(move || { + if let Err(error) = handle_connection(stream, &handler_state) { + handler_state.handler_failed.store(true, Ordering::SeqCst); + eprintln!("test registry handler failed: {error:#}"); + } + }); + } + }) + .context("starting test registry accept loop")?; + Ok(Self { + address, + state, + shutdown, + accept_thread: Some(accept_thread), + }) + } + + fn base_url(&self) -> String { + format!("http://{}", self.address) + } + + fn max_active_artifacts(&self) -> usize { + self.state.max_active_artifacts.load(Ordering::SeqCst) + } + + fn artifact_requests(&self) -> usize { + self.state.artifact_requests.load(Ordering::SeqCst) + } + + fn gate_timed_out(&self) -> bool { + self.state + .gate + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .timed_out + } +} + +impl Drop for RegistryServer { + fn drop(&mut self) { + self.shutdown.store(true, Ordering::SeqCst); + let _ = TcpStream::connect(self.address); + if let Some(thread) = self.accept_thread.take() { + let _ = thread.join(); + } + } +} + +fn handle_connection(mut stream: TcpStream, state: &ServerState) -> Result<()> { + let mut reader = BufReader::new(stream.try_clone()?); + let mut request_line = String::new(); + if reader.read_line(&mut request_line)? == 0 { + return Ok(()); + } + let path = request_line + .split_whitespace() + .nth(1) + .context("HTTP request has a path")? + .split('?') + .next() + .unwrap_or_default() + .to_string(); + + loop { + let mut header = String::new(); + if reader.read_line(&mut header)? == 0 || header == "\r\n" || header == "\n" { + break; + } + } + + let Some(response) = state.responses.get(&path) else { + return write_response(&mut stream, "404 Not Found", "text/plain", b"not found\n"); + }; + + if !response.artifact { + return write_response( + &mut stream, + "200 OK", + response.content_type, + response.body.as_slice(), + ); + } + + state.artifact_requests.fetch_add(1, Ordering::SeqCst); + let active = state.active_artifacts.fetch_add(1, Ordering::SeqCst) + 1; + state + .max_active_artifacts + .fetch_max(active, Ordering::SeqCst); + + let mut gate = state + .gate + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + gate.arrived += 1; + if gate.arrived >= EXPECTED_CONCURRENCY { + gate.open = true; + state.gate_ready.notify_all(); + } + let deadline = Instant::now() + Duration::from_secs(5); + while !gate.open { + let remaining = deadline.saturating_duration_since(Instant::now()); + if remaining.is_zero() { + gate.timed_out = true; + gate.open = true; + state.gate_ready.notify_all(); + break; + } + let (next, timeout) = state + .gate_ready + .wait_timeout(gate, remaining) + .unwrap_or_else(|poisoned| poisoned.into_inner()); + gate = next; + if timeout.timed_out() && !gate.open { + gate.timed_out = true; + gate.open = true; + state.gate_ready.notify_all(); + } + } + drop(gate); + + // Keep the first wave alive briefly so an implementation that exceeds the + // configured bound is visible in max_active_artifacts rather than hidden + // by very small fixture archives. + thread::sleep(Duration::from_millis(75)); + let result = write_response( + &mut stream, + "200 OK", + response.content_type, + response.body.as_slice(), + ); + state.active_artifacts.fetch_sub(1, Ordering::SeqCst); + result +} + +fn write_response( + stream: &mut TcpStream, + status: &str, + content_type: &str, + body: &[u8], +) -> Result<()> { + write!( + stream, + "HTTP/1.1 {status}\r\nContent-Type: {content_type}\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", + body.len() + )?; + stream.write_all(body)?; + stream.flush()?; + Ok(()) +} + +struct PublishedPackage { + name: String, + package: PackageMetadata, + version: VersionMetadata, + artifact: Vec, +} + +fn package_manifest(name: &str) -> String { + format!( + r#"[package] +org = "{ORG}" +name = "{name}" +version = "{VERSION}" +description = "bounded recursive install fixture" + +[package.repository] +vcs = "git" +url = "https://example.invalid/{ORG}/{name}" +"#, + ) +} + +fn build_package(scratch: &Path, name: &str) -> Result { + let source = scratch.join(format!("source-{name}")); + fs::create_dir_all(&source)?; + let manifest_text = package_manifest(name); + fs::write(source.join(".zpkg.toml"), &manifest_text)?; + fs::write( + source.join("payload.txt"), + format!("{ORG}/{name}@{VERSION}\n"), + )?; + let manifest = Manifest::parse(&manifest_text)?; + let packed = pack( + &source, + &manifest, + Some(&scratch.join(format!("packed-{name}"))), + )?; + let artifact = fs::read(&packed.path)?; + Ok(PublishedPackage { + name: name.to_string(), + package: PackageMetadata { + org: ORG.to_string(), + name: name.to_string(), + description: Some(format!("fixture {name}")), + vcs: Vcs::Git, + repo_url: format!("https://example.invalid/{ORG}/{name}"), + version_scheme: manifest.package.version_scheme, + latest: Some(VERSION.to_string()), + tags: Vec::new(), + versions: vec![VERSION.to_string()], + }, + version: VersionMetadata { + org: ORG.to_string(), + name: name.to_string(), + version: VERSION.to_string(), + sha256: packed.sha256, + size: packed.size, + format: packed.format, + vcs_tag: format!("v{VERSION}"), + vcs_commit: Some("0123456789abcdef0123456789abcdef01234567".to_string()), + download_url: String::new(), + published_at: "1970-01-01T00:00:00Z".to_string(), + yanked: false, + }, + artifact, + }) +} + +fn registry_responses( + address: SocketAddr, + packages: &mut [PublishedPackage], +) -> Result> { + let mut responses = HashMap::new(); + for package in packages { + let artifact_path = registry::artifact_path(&package.version.sha256); + package.version.download_url = format!("http://{address}{artifact_path}"); + responses.insert( + registry::package_path(ORG, &package.name), + Response { + content_type: "application/json", + body: Arc::new(serde_json::to_vec(&package.package)?), + artifact: false, + }, + ); + responses.insert( + registry::version_path(ORG, &package.name, VERSION), + Response { + content_type: "application/json", + body: Arc::new(serde_json::to_vec(&package.version)?), + artifact: false, + }, + ); + responses.insert( + artifact_path, + Response { + content_type: "application/octet-stream", + body: Arc::new(package.artifact.clone()), + artifact: true, + }, + ); + } + Ok(responses) +} + +fn write_consumer(project: &Path, packages: &[PublishedPackage]) -> Result<()> { + fs::create_dir_all(project)?; + let mut manifest = String::from( + r#"[package] +org = "recursive-consumer" +name = "wide-frontier" +version = "0.1.0" + +[package.repository] +vcs = "git" +url = "https://example.invalid/recursive-consumer/wide-frontier" + +[dependencies] +"#, + ); + for package in packages { + manifest.push_str(&format!("\"{ORG}/{}\" = \"={VERSION}\"\n", package.name)); + } + fs::write(project.join(".zpkg.toml"), manifest)?; + Ok(()) +} + +fn test_config(registry: &str, home: PathBuf) -> Config { + Config { + registry: registry.to_string(), + home, + token: None, + auth_url: "http://127.0.0.1/unused".to_string(), + supabase_url: None, + supabase_key: None, + interactive: false, + } +} + +#[test] +fn recursive_http_prefetch_saturates_at_five_and_warm_runs_do_not_redownload() -> Result<()> { + if let Ok(raw) = std::env::var("ZED_PKG_INSTALL_CONCURRENCY") { + assert_eq!( + raw, "5", + "this contract test expects the default five-worker setting or an explicit value of 5" + ); + } + + let temp = tempfile::tempdir()?; + let scratch = temp.path().join("scratch"); + let project = temp.path().join("project"); + let home = temp.path().join("home"); + let listener = TcpListener::bind("127.0.0.1:0")?; + let address = listener.local_addr()?; + + let mut packages = (0..PACKAGE_COUNT) + .map(|index| build_package(&scratch, &format!("pkg-{index:02}"))) + .collect::>>()?; + write_consumer(&project, &packages)?; + let server = RegistryServer::start(listener, registry_responses(address, &mut packages)?)?; + let config = test_config(&server.base_url(), home.clone()); + + let cold = prefetch(&project, &config, false)?; + assert_eq!(cold.resolved, PACKAGE_COUNT); + assert_eq!(cold.downloaded, PACKAGE_COUNT); + assert_eq!(server.artifact_requests(), PACKAGE_COUNT); + assert_eq!(server.max_active_artifacts(), EXPECTED_CONCURRENCY); + assert!( + !server.gate_timed_out(), + "five artifact handlers never became active together" + ); + assert!( + !server.state.handler_failed.load(Ordering::SeqCst), + "the fixture HTTP registry reported a handler failure" + ); + + let warm = prefetch(&project, &config, false)?; + assert_eq!(warm.resolved, PACKAGE_COUNT); + assert_eq!(warm.downloaded, 0); + assert_eq!( + server.artifact_requests(), + PACKAGE_COUNT, + "a warm content-addressed store must not fetch artifact bytes again" + ); + + let store = Store::new(&home); + for package in &packages { + assert!( + store.has(&package.version.sha256), + "missing store entry for {}/{}", + package.version.org, + package.version.name + ); + } + + let artifact_locks = fs::read_dir(home.join("locks"))? + .filter_map(|entry| entry.ok()) + .filter(|entry| entry.file_name().to_string_lossy().starts_with("artifact-")) + .count(); + assert_eq!(artifact_locks, PACKAGE_COUNT); + + let cache_entries = fs::read_dir(home.join("cache"))?.collect::>>()?; + assert_eq!(cache_entries.len(), PACKAGE_COUNT); + assert!( + cache_entries + .iter() + .all(|entry| entry.file_type().is_ok_and(|kind| kind.is_file())), + "atomic download staging directories must be removed" + ); + Ok(()) +}