Skip to content

fix: serialize configure locator updates #416

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
karthiknadig wants to merge 5 commits into
base: main
Choose a base branch
from
+369 −46
Open

fix: serialize configure locator updates #416

Changes from 2 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
3337ff1
fix: serialize configure locator updates (Fixes #385)
karthiknadig Apr 12, 2026
fe7c683
fix: address configure review feedback (PR #416)
karthiknadig Apr 12, 2026
8e5d3f9
fix: harden configure failure handling (PR #416)
karthiknadig Apr 12, 2026
ee78c4c
test: stabilize virtualenvwrapper project path on macOS (PR #416)
karthiknadig Apr 12, 2026
0b29cb1
fix: harden configure rollback semantics (PR #416)
karthiknadig Apr 12, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
190 changes: 146 additions & 44 deletions crates/pet/src/jsonrpc.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -531,41 +531,45 @@ const GLOB_EXPANSION_WARN_THRESHOLD: Duration = Duration::from_secs(5);

pub fn handle_configure(context: Arc<Context>, id: u32, params: Value) {
match serde_json::from_value::<ConfigureOptions>(params.clone()) {
Ok(configure_options) => {
Ok(mut configure_options) => {
info!("Received configure request");
// Start in a new thread, we can have multiple requests.
thread::spawn(move || {
let now = Instant::now();

// Expand glob patterns before acquiring the write lock so we
// don't block readers/writers while traversing the filesystem.
let workspace_directories = configure_options.workspace_directories.map(|dirs| {
let start = Instant::now();
let result: Vec<PathBuf> = expand_glob_patterns(&dirs)
.into_iter()
.filter(|p| p.is_dir())
.collect();
trace!(
"Expanded workspace directory patterns ({:?}) in {:?}",
dirs,
start.elapsed()
);
result
});
let environment_directories =
configure_options.environment_directories.map(|dirs| {
let workspace_directories =
configure_options.workspace_directories.take().map(|dirs| {
let start = Instant::now();
let result: Vec<PathBuf> = expand_glob_patterns(&dirs)
.into_iter()
.filter(|p| p.is_dir())
.collect();
trace!(
"Expanded environment directory patterns ({:?}) in {:?}",
"Expanded workspace directory patterns ({:?}) in {:?}",
dirs,
start.elapsed()
);
result
});
let environment_directories =
configure_options
.environment_directories
.take()
.map(|dirs| {
let start = Instant::now();
let result: Vec<PathBuf> = expand_glob_patterns(&dirs)
.into_iter()
.filter(|p| p.is_dir())
.collect();
trace!(
"Expanded environment directory patterns ({:?}) in {:?}",
dirs,
start.elapsed()
);
result
});
let glob_elapsed = now.elapsed();
trace!("Glob expansion completed in {:?}", glob_elapsed);
if glob_elapsed >= GLOB_EXPANSION_WARN_THRESHOLD {
Expand All @@ -575,33 +579,13 @@ pub fn handle_configure(context: Arc<Context>, id: u32, params: Value) {
);
}

let config = {
let mut state = context.configuration.write().unwrap();
state.config.workspace_directories = workspace_directories;
state.config.conda_executable = configure_options.conda_executable;
state.config.environment_directories = environment_directories;
state.config.pipenv_executable = configure_options.pipenv_executable;
state.config.poetry_executable = configure_options.poetry_executable;
// We will not support changing the cache directories once set.
// No point, supporting such a use case.
if let Some(cache_directory) = configure_options.cache_directory {
set_cache_directory(cache_directory.clone());
state.config.cache_directory = Some(cache_directory);
}
state.generation += 1;
// Reset missing-env reporting so that the next refresh
// after reconfiguration can trigger it again (Fixes #395).
// Done inside the write lock to avoid a TOCTOU window with
// concurrent refresh threads reading the generation.
MISSING_ENVS_REPORTING_STATE.store(MISSING_ENVS_AVAILABLE, Ordering::Release);
trace!(
"Configuring locators with generation {}: {:?}",
state.generation,
state.config
);
state.config.clone()
};
configure_locators(&context.locators, &config);
apply_configure_options(
context.configuration.as_ref(),
&context.locators,
configure_options,
workspace_directories,
environment_directories,
);
info!("Configure completed in {:?}", now.elapsed());
send_reply(id, None::<()>);
});
Expand Down Expand Up @@ -665,6 +649,39 @@ fn parse_refresh_options(params: Value) -> Result<RefreshOptions, serde_json::Er
.map(|options| canonicalize_refresh_options(options.unwrap_or_default()))
}

fn apply_configure_options(
configuration: &RwLock<ConfigurationState>,
locators: &Arc<Vec<Arc<dyn Locator>>>,
configure_options: ConfigureOptions,
workspace_directories: Option<Vec<PathBuf>>,
environment_directories: Option<Vec<PathBuf>>,
) {
let mut state = configuration.write().unwrap();
state.config.workspace_directories = workspace_directories;
state.config.conda_executable = configure_options.conda_executable;
state.config.environment_directories = environment_directories;
state.config.pipenv_executable = configure_options.pipenv_executable;
state.config.poetry_executable = configure_options.poetry_executable;
// We will not support changing the cache directories once set.
// No point, supporting such a use case.
if let Some(cache_directory) = configure_options.cache_directory {
set_cache_directory(cache_directory.clone());
state.config.cache_directory = Some(cache_directory);
}
state.generation += 1;
// Reset missing-env reporting so that the next refresh after
// reconfiguration can trigger it again (Fixes #395). Done inside the write
// lock to avoid a TOCTOU window with concurrent refresh threads reading the
// generation.
MISSING_ENVS_REPORTING_STATE.store(MISSING_ENVS_AVAILABLE, Ordering::Release);
trace!(
"Configuring locators with generation {}: {:?}",
state.generation,
state.config
);
configure_locators(locators, &state.config);
}

fn configure_locators(locators: &Arc<Vec<Arc<dyn Locator>>>, config: &Configuration) {
for locator in locators.iter() {
locator.configure(config);
Expand Down Expand Up @@ -1244,6 +1261,12 @@ mod tests {
reported: Mutex<bool>,
}

struct BlockingConfigureLocator {
started: mpsc::Sender<()>,
release: Mutex<mpsc::Receiver<()>>,
configured_workspace_directories: Mutex<Option<Vec<PathBuf>>>,
}

impl Reporter for RecordingReporter {
fn report_manager(&self, manager: &EnvManager) {
self.managers.lock().unwrap().push(manager.clone());
Expand Down Expand Up @@ -1275,6 +1298,29 @@ mod tests {
}
}

impl Locator for BlockingConfigureLocator {
fn get_kind(&self) -> LocatorKind {
LocatorKind::Venv
}

fn configure(&self, config: &Configuration) {
self.started.send(()).unwrap();
self.release.lock().unwrap().recv().unwrap();
*self.configured_workspace_directories.lock().unwrap() =
config.workspace_directories.clone();
}

fn supported_categories(&self) -> Vec<PythonEnvironmentKind> {
vec![PythonEnvironmentKind::Venv]
}

fn try_from(&self, _env: &pet_core::env::PythonEnv) -> Option<PythonEnvironment> {
None
}

fn find(&self, _reporter: &dyn Reporter) {}
}

fn make_refresh_key(generation: u64, options: RefreshOptions) -> RefreshKey {
RefreshKey::new(&options, generation)
}
Expand Down Expand Up @@ -1402,6 +1448,62 @@ mod tests {
assert!(*inner.reported.lock().unwrap());
}

#[test]
fn test_configure_publishes_state_after_shared_locators_are_configured() {
let configuration = Arc::new(RwLock::new(ConfigurationState::default()));
let (started_tx, started_rx) = mpsc::channel();
let (release_tx, release_rx) = mpsc::channel();
let (done_tx, done_rx) = mpsc::channel();
let locator = Arc::new(BlockingConfigureLocator {
started: started_tx,
release: Mutex::new(release_rx),
configured_workspace_directories: Mutex::new(None),
});
let locators = Arc::new(vec![locator.clone() as Arc<dyn Locator>]);
let workspace_directories = vec![PathBuf::from("/workspace")];

let worker = {
let configuration = configuration.clone();
let locators = locators.clone();
let workspace_directories = workspace_directories.clone();
thread::spawn(move || {
apply_configure_options(
configuration.as_ref(),
&locators,
ConfigureOptions {
workspace_directories: None,
conda_executable: None,
pipenv_executable: None,
poetry_executable: None,
environment_directories: None,
cache_directory: None,
},
Some(workspace_directories),
None,
);
done_tx.send(()).unwrap();
})
};

started_rx.recv_timeout(Duration::from_secs(5)).unwrap();
assert!(configuration.try_read().is_err());

release_tx.send(()).unwrap();
done_rx.recv_timeout(Duration::from_secs(5)).unwrap();
worker.join().unwrap();

let state = configuration.read().unwrap();
assert_eq!(state.generation, 1);
assert_eq!(
state.config.workspace_directories,
Some(workspace_directories.clone())
);
assert_eq!(
*locator.configured_workspace_directories.lock().unwrap(),
Some(workspace_directories)
);
}

#[test]
fn test_stale_generation_does_not_begin_missing_env_reporting() {
let _guard = MISSING_ENVS_TEST_LOCK.lock().unwrap();
Expand Down
Loading