perf(pet-windows-registry): parallelize HKLM/HKCU walk and reuse cache across refresh (#458)

Fixes #454.

The `WindowsRegistry` locator walked HKLM and HKCU **serially** on every
`refresh`, with no per-company parallelism either, despite `pet-pyenv` /
`pet-homebrew` / `pet-conda` already establishing the `thread::scope`
pattern. On Windows hosts with Defender intercepting every registry
read, this was a measurable contributor to the steady-state p90 refresh
latency (#454).

## Changes

- **Parallelize HKLM and HKCU** using `thread::scope` (pyenv pattern).
Each hive runs in its own scoped thread; results are stitched back in a
deterministic HKLM-then-HKCU order.
- **Parallelize per-company within each hive.** Companies are enumerated
and their subkeys opened serially (cheap handle ops), then each is moved
into its own scoped thread. `winreg::RegKey: Send + !Sync`, so each
thread owns its handle by value — no `&RegKey` crosses a thread
boundary.
- **Drop the `self.clear()` at the top of `find()`.** `find_with_cache`
already short-circuits on a cache hit, and `find()` runs on transient
locators per refresh (the cache is empty by construction the first
time), so the unconditional clear was forcing every `refresh` RPC to pay
full registry-walk cost. Makes `find()` idempotent.
- **Don't silently swallow panics in spawned threads.** New
`join_or_warn(...)` helper logs the panic payload via `warn!` before
substituting `empty_result()` so a thread crash in one hive/company
doesn't make the entire walk silently disappear.

## Tests

- `test_find_reuses_cached_results_within_locator_lifetime` —
pre-populates the cache, runs `find()`, asserts entries survived (pins
down the `clear()` removal).
- `test_find_on_fresh_locator_completes_parallel_walk` — runs `find()`
on an empty cache and asserts the cache becomes `Some(_)`. Smoke-test
for the new parallel scaffolding; doesn't depend on registry contents so
passes on CI hosts without Python installs.

## Out of scope

The "split `WindowsRegistry` timing into HKLM vs HKCU in
`RefreshPerformance`" telemetry follow-up mentioned in #454 is a
separate change; happy to file as its own issue if useful.

Author: karthiknadig

crates/pet-windows-registry/src/environments.rs

@@ -8,7 +8,6 @@ use pet_core::reporter::Reporter;
 #[cfg(windows)]
 use pet_core::{
     arch::Architecture,
-    manager::EnvManager,
     python_environment::{PythonEnvironmentBuilder, PythonEnvironmentKind},
     LocatorResult,
 };
@@ -19,64 +18,250 @@ use std::{path::PathBuf, sync::Arc};
 #[cfg(windows)]
 use winreg::RegKey;
 
+#[cfg(windows)]
+fn empty_result() -> LocatorResult {
+    LocatorResult {
+        environments: vec![],
+        managers: vec![],
+    }
+}
+
+/// What a single company-level worker thread produces: the registry-derived
+/// `PythonEnvironment`s plus the install dirs that turned out to be conda
+/// roots. We capture conda dirs (instead of just calling
+/// `conda_locator.find_and_report` and forgetting) so that subsequent
+/// `find()` calls on a cache hit can replay them — otherwise registry-only
+/// conda installs would silently disappear after the first refresh.
+#[cfg(windows)]
+struct CompanyWalkOutcome {
+    result: LocatorResult,
+    conda_install_dirs: Vec<PathBuf>,
+}
+
+#[cfg(windows)]
+impl CompanyWalkOutcome {
+    fn empty() -> Self {
+        Self {
+            result: empty_result(),
+            conda_install_dirs: vec![],
+        }
+    }
+}
+
+/// Logs a warning if a spawned registry-walk thread panicked, then
+/// substitutes an empty outcome so the surviving hive/companies still
+/// surface their environments. Without this we'd silently lose the entire
+/// hive when one company's walk panics — exactly the kind of regression
+/// that's hardest to debug after the fact.
+///
+/// Returns `(outcome, had_panic)` so callers can decide not to cache a
+/// partial result (otherwise a single transient panic would persist as a
+/// stale empty/partial cache across refreshes — see issue #454).
+#[cfg(windows)]
+fn join_or_warn(
+    join_result: std::thread::Result<CompanyWalkOutcome>,
+    label: &str,
+) -> (CompanyWalkOutcome, bool) {
+    use log::warn;
+    match join_result {
+        Ok(outcome) => (outcome, false),
+        Err(panic_payload) => {
+            // Try to render the payload for the log; payloads are commonly
+            // either a `&'static str` or a `String`.
+            let message = panic_payload
+                .downcast_ref::<&'static str>()
+                .map(|s| (*s).to_string())
+                .or_else(|| panic_payload.downcast_ref::<String>().cloned())
+                .unwrap_or_else(|| "<non-string panic payload>".to_string());
+            warn!("Registry walk thread for {} panicked: {}", label, message);
+            (CompanyWalkOutcome::empty(), true)
+        }
+    }
+}
+
+/// Outcome of a registry walk: the discovered environments/managers, the
+/// conda install dirs found via the registry (so cache-hit replays can
+/// re-invoke `conda_locator.find_and_report`), and a flag indicating
+/// whether any worker thread panicked. Callers should avoid caching a
+/// result with `had_panic = true` so a transient failure can be retried
+/// on the next refresh instead of becoming sticky.
+#[cfg(windows)]
+pub struct RegistryWalkOutcome {
+    pub result: LocatorResult,
+    pub conda_install_dirs: Vec<PathBuf>,
+    pub had_panic: bool,
+}
+
 #[cfg(windows)]
 pub fn get_registry_pythons(
     conda_locator: &Arc<dyn CondaLocator>,
     reporter: &Option<&dyn Reporter>,
-) -> LocatorResult {
-    use log::{trace, warn};
+) -> RegistryWalkOutcome {
+    use std::thread;
 
-    let mut environments = vec![];
-    let mut managers: Vec<EnvManager> = vec![];
+    // Walk both hives in parallel. Each hive walks its companies in parallel
+    // too (see `get_registry_pythons_for_hive`). HKLM and HKCU sit on
+    // independent registry trees and Defender intercepts every read, so the
+    // serial baseline was paying for both round-trips back to back; the
+    // scope-spawn pattern matches `pet-pyenv` / `pet-homebrew` / `pet-conda`.
+    let (hklm_outcome, hkcu_outcome) = thread::scope(|s| {
+        let hklm = s.spawn(|| {
+            get_registry_pythons_for_hive(
+                "HKLM",
+                RegKey::predef(winreg::enums::HKEY_LOCAL_MACHINE),
+                conda_locator,
+                reporter,
+            )
+        });
+        let hkcu = s.spawn(|| {
+            get_registry_pythons_for_hive(
+                "HKCU",
+                RegKey::predef(winreg::enums::HKEY_CURRENT_USER),
+                conda_locator,
+                reporter,
+            )
+        });
+        (
+            join_hive_outcome(hklm.join(), "HKLM"),
+            join_hive_outcome(hkcu.join(), "HKCU"),
+        )
+    });
 
-    struct RegistryKey {
-        pub name: &'static str,
-        pub key: winreg::RegKey,
-    }
-    let search_keys = [
-        RegistryKey {
-            name: "HKLM",
-            key: winreg::RegKey::predef(winreg::enums::HKEY_LOCAL_MACHINE),
-        },
-        RegistryKey {
-            name: "HKCU",
-            key: winreg::RegKey::predef(winreg::enums::HKEY_CURRENT_USER),
+    let mut environments = hklm_outcome.result.environments;
+    environments.extend(hkcu_outcome.result.environments);
+    let mut managers = hklm_outcome.result.managers;
+    managers.extend(hkcu_outcome.result.managers);
+    let mut conda_install_dirs = hklm_outcome.conda_install_dirs;
+    conda_install_dirs.extend(hkcu_outcome.conda_install_dirs);
+
+    RegistryWalkOutcome {
+        result: LocatorResult {
+            environments,
+            managers,
         },
-    ];
-    for (name, key) in search_keys.iter().map(|f| (f.name, &f.key)) {
-        match key.open_subkey("Software\\Python") {
-            Ok(python_key) => {
-                for company in python_key.enum_keys().filter_map(Result::ok) {
-                    trace!("Searching {}\\Software\\Python\\{}", name, company);
-                    match python_key.open_subkey(&company) {
-                        Ok(company_key) => {
-                            let result = get_registry_pythons_from_key_for_company(
-                                name,
-                                &company_key,
-                                &company,
-                                conda_locator,
-                                reporter,
-                            );
-                            managers.extend(result.managers);
-                            environments.extend(result.environments);
-                        }
-                        Err(err) => {
-                            warn!(
-                                "Failed to open {}\\Software\\Python\\{}, {:?}",
-                                name, company, err
-                            );
-                        }
-                    }
-                }
+        conda_install_dirs,
+        had_panic: hklm_outcome.had_panic || hkcu_outcome.had_panic,
+    }
+}
+
+/// Sibling of `join_or_warn` for hive-level threads. Returns the recovered
+/// outcome (already aggregating any company-level panics) so the top
+/// level can OR the panic flags and concatenate the conda install dirs.
+#[cfg(windows)]
+fn join_hive_outcome(
+    join_result: std::thread::Result<RegistryWalkOutcome>,
+    label: &str,
+) -> RegistryWalkOutcome {
+    use log::warn;
+    match join_result {
+        Ok(outcome) => outcome,
+        Err(panic_payload) => {
+            let message = panic_payload
+                .downcast_ref::<&'static str>()
+                .map(|s| (*s).to_string())
+                .or_else(|| panic_payload.downcast_ref::<String>().cloned())
+                .unwrap_or_else(|| "<non-string panic payload>".to_string());
+            warn!("Registry walk thread for {} panicked: {}", label, message);
+            RegistryWalkOutcome {
+                result: empty_result(),
+                conda_install_dirs: vec![],
+                had_panic: true,
             }
+        }
+    }
+}
+
+/// Walks `<hive>\Software\Python\<company>` for every company in the given
+/// hive. Companies are processed in parallel; each spawned thread owns its
+/// own `RegKey` handle (which is `Send` but not `Sync` in `winreg`).
+#[cfg(windows)]
+fn get_registry_pythons_for_hive(
+    name: &'static str,
+    hive: RegKey,
+    conda_locator: &Arc<dyn CondaLocator>,
+    reporter: &Option<&dyn Reporter>,
+) -> RegistryWalkOutcome {
+    use log::{trace, warn};
+    use std::thread;
+
+    let python_key = match hive.open_subkey("Software\\Python") {
+        Ok(k) => k,
+        Err(err) => {
+            warn!("Failed to open {}\\Software\\Python, {:?}", name, err);
+            return RegistryWalkOutcome {
+                result: empty_result(),
+                conda_install_dirs: vec![],
+                had_panic: false,
+            };
+        }
+    };
+
+    // Open each company subkey serially. Opening a registry handle is cheap
+    // (no recursive enumeration); the heavy work happens once we start
+    // pulling values out of `<company>\<install>\InstallPath`. Collecting
+    // owned `(String, RegKey)` pairs lets us hand each company to its own
+    // thread without sharing a `RegKey` (which is `Send` but not `Sync`).
+    let companies: Vec<(String, RegKey)> = python_key
+        .enum_keys()
+        .filter_map(Result::ok)
+        .filter_map(|company| match python_key.open_subkey(&company) {
+            Ok(company_key) => Some((company, company_key)),
             Err(err) => {
-                warn!("Failed to open {}\\Software\\Python, {:?}", name, err)
+                warn!(
+                    "Failed to open {}\\Software\\Python\\{}, {:?}",
+                    name, company, err
+                );
+                None
             }
-        }
+        })
+        .collect();
+
+    let results: Vec<(CompanyWalkOutcome, bool)> = thread::scope(|s| {
+        let handles: Vec<_> = companies
+            .into_iter()
+            .map(|(company, company_key)| {
+                // Build the panic-warning label up-front so a panicking
+                // company thread is identifiable in logs (issue #454).
+                let label = format!("{name}\\Software\\Python\\{company}");
+                let handle = s.spawn(move || {
+                    // Trace order is intentionally relaxed: companies are
+                    // walked in parallel, so this line interleaves with the
+                    // others from the same hive.
+                    trace!("Searching {}\\Software\\Python\\{}", name, company);
+                    get_registry_pythons_from_key_for_company(
+                        name,
+                        &company_key,
+                        &company,
+                        conda_locator,
+                        reporter,
+                    )
+                });
+                (label, handle)
+            })
+            .collect();
+        handles
+            .into_iter()
+            .map(|(label, h)| join_or_warn(h.join(), &label))
+            .collect()
+    });
+
+    let mut environments = vec![];
+    let mut managers = vec![];
+    let mut conda_install_dirs = vec![];
+    let mut had_panic = false;
+    for (outcome, panicked) in results {
+        environments.extend(outcome.result.environments);
+        managers.extend(outcome.result.managers);
+        conda_install_dirs.extend(outcome.conda_install_dirs);
+        had_panic |= panicked;
     }
-    LocatorResult {
-        environments,
-        managers,
+    RegistryWalkOutcome {
+        result: LocatorResult {
+            environments,
+            managers,
+        },
+        conda_install_dirs,
+        had_panic,
     }
 }
 
@@ -87,12 +272,13 @@ fn get_registry_pythons_from_key_for_company(
     company: &str,
     conda_locator: &Arc<dyn CondaLocator>,
     reporter: &Option<&dyn Reporter>,
-) -> LocatorResult {
+) -> CompanyWalkOutcome {
     use log::{trace, warn};
     use pet_conda::utils::is_conda_env;
     use pet_fs::path::norm_case;
 
     let mut environments = vec![];
+    let mut conda_install_dirs = vec![];
     // let company_display_name: Option<String> = company_key.get_value("DisplayName").ok();
     for installed_python in company_key.enum_keys().filter_map(Result::ok) {
         match company_key.open_subkey(installed_python.clone()) {
@@ -132,6 +318,15 @@ fn get_registry_pythons_from_key_for_company(
                             if let Some(reporter) = reporter {
                                 conda_locator.find_and_report(*reporter, &env_path);
                             }
+                            // Capture the dir even when no reporter is
+                            // attached (e.g. cache-warming via
+                            // `find_with_cache(None)`) so a later cache
+                            // hit in `find()` can replay
+                            // `conda_locator.find_and_report` against
+                            // each dir; without this, registry-only conda
+                            // installs would silently disappear from
+                            // every refresh after the first (#454).
+                            conda_install_dirs.push(env_path);
                             continue;
                         }
 
@@ -217,8 +412,11 @@ fn get_registry_pythons_from_key_for_company(
         }
     }
 
-    LocatorResult {
-        environments,
-        managers: vec![],
+    CompanyWalkOutcome {
+        result: LocatorResult {
+            environments,
+            managers: vec![],
+        },
+        conda_install_dirs,
     }
 }