Skip to content

refactor: Port Cargo watch and prune knowledge #13582

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

Merged
anthonyshew merged 1 commit into from
Jul 30, 2026
Merged

refactor: Port Cargo watch and prune knowledge #13582

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
28 changes: 0 additions & 28 deletions crates/turborepo-engine/src/builder/test.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -140,20 +140,6 @@ impl Toolchain for AggregateToolchain {
))
})
}

fn watch_spec(&self) -> turborepo_repository::toolchain::WatchSpec {
turborepo_repository::toolchain::WatchSpec::default()
}

fn prune_plan(
&self,
_kept_packages: &[String],
) -> Result<
Option<turborepo_repository::toolchain::PrunePlan>,
turborepo_repository::toolchain::Error,
> {
Ok(None)
}
}

type StubIOEngineResult = Engine<Built, TaskDefinition>;
Expand Down Expand Up @@ -218,20 +204,6 @@ impl Toolchain for StubIOToolchain {
))
})
}

fn watch_spec(&self) -> turborepo_repository::toolchain::WatchSpec {
turborepo_repository::toolchain::WatchSpec::default()
}

fn prune_plan(
&self,
_kept_packages: &[String],
) -> Result<
Option<turborepo_repository::toolchain::PrunePlan>,
turborepo_repository::toolchain::Error,
> {
Ok(None)
}
}

fn stub_io_package_graph(
Expand Down
16 changes: 8 additions & 8 deletions crates/turborepo-lib/src/commands/prune.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -85,6 +85,8 @@ pub enum Error {
PackageNotPruneable(String),
#[error(transparent)]
Toolchain(#[from] turborepo_repository::toolchain::Error),
#[error(transparent)]
PruneKnowledge(#[from] turborepo_repository::prune_knowledge::Error),
}

static ADDITIONAL_FILES: LazyLock<Vec<(&'static RelativeUnixPath, Option<CopyDestination>)>> =
Expand Down Expand Up @@ -289,16 +291,14 @@ pub async fn prune(
}
}

// Each toolchain contributes whatever the pruned repository needs
// beyond the packages themselves: extra members it requires, rewritten
// workspace files, and config files to carry over.
for toolchain in prune.package_graph.toolchains().iter() {
let toolchain_id = toolchain.id();
let kept = kept_by_toolchain.remove(&toolchain_id).unwrap_or_default();
let Some(plan) = toolchain.prune_plan(&kept)? else {
// Project plans from immutable knowledge captured by this graph's
// discovery generation; live toolchains retain no prune authority.
for toolchain_id in prune.package_graph.prune_toolchains() {
let kept = kept_by_toolchain.remove(toolchain_id).unwrap_or_default();
let Some(plan) = prune.package_graph.prune_plan(toolchain_id, &kept)? else {
continue;
};
planned_toolchains.insert(toolchain_id);
planned_toolchains.insert(toolchain_id.clone());
for extra in plan.extra_packages {
let name = PackageName::Other(extra.clone());
let context = prune.package_context(&name)?;
Expand Down
91 changes: 62 additions & 29 deletions crates/turborepo-lib/src/package_changes_watcher.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,10 @@ use std::{
collections::{HashMap, HashSet},
io::ErrorKind,
ops::DerefMut,
sync::{Arc, RwLock},
sync::{
atomic::{AtomicBool, Ordering},
Arc, RwLock,
},
};

use notify::Event;
Expand Down Expand Up @@ -125,6 +128,7 @@ struct Subscriber {
repo_root: AbsoluteSystemPathBuf,
repository_ignore: RepositoryIgnore,
watch_spec: Arc<RwLock<WatchSpec>>,
watch_spec_ready: Arc<AtomicBool>,
package_change_events_tx: broadcast::Sender<PackageChangeEvent>,
hash_watcher: Arc<HashWatcher>,
custom_turbo_json_path: Option<AbsoluteSystemPathBuf>,
Expand Down Expand Up @@ -251,11 +255,10 @@ fn classify_changed_files(
// Whether an anchored path is under one of the toolchains'
// build-byproduct directories.
let in_ignored_prefix = |path: &AnchoredSystemPathBuf| {
watch_spec.ignore_prefixes.iter().any(|prefix| {
path.components()
.next()
.is_some_and(|component| component.as_str() == prefix)
})
watch_spec
.ignore_prefixes
.iter()
.any(|prefix| path_is_under_prefix(path, prefix))
};

// Toolchain workspace-definition files (e.g. Cargo manifests and the
Expand Down Expand Up @@ -330,6 +333,15 @@ fn classify_changed_files(
}
}

fn path_is_under_prefix(path: &AnchoredSystemPath, prefix: &str) -> bool {
let path = path.to_unix();
path.as_str() == prefix
|| path
.as_str()
.strip_prefix(prefix)
.is_some_and(|suffix| suffix.starts_with('/'))
}

impl RepoState {
fn get_change_mapper(&self) -> Option<ChangeMapper<'_, GlobalDepsPackageChangeMapper<'_>>> {
let Ok(package_change_mapper) = GlobalDepsPackageChangeMapper::new(
Expand Down Expand Up @@ -398,19 +410,16 @@ impl Subscriber {
let repository_ignore = file_events
.repository_ignore()
.unwrap_or_else(|| RepositoryIgnore::new(repo_root.as_std_path()));
let mut watch_spec = WatchSpec::default();
if !single_package {
for toolchain in &extra_toolchains {
watch_spec.extend(toolchain.watch_spec());
}
}

Subscriber {
repo_root,
file_events,
changed_files: Default::default(),
repository_ignore,
watch_spec: Arc::new(RwLock::new(watch_spec)),
watch_spec: Arc::new(RwLock::new(WatchSpec::default())),
// Before the first graph generation is published, retain every
// in-repository event. This closes the discovery/subscription race
// without consulting live toolchains for bootstrap facts.
watch_spec_ready: Arc::new(AtomicBool::new(false)),
package_change_events_tx,
hash_watcher,
custom_turbo_json_path: normalized_custom_path,
Expand Down Expand Up @@ -506,6 +515,7 @@ impl Subscriber {
.watch_spec
.write()
.unwrap_or_else(|poisoned| poisoned.into_inner()) = pkg_dep_graph.active_watch_spec();
self.watch_spec_ready.store(true, Ordering::Release);

Some(RepoState {
root_turbo_json,
Expand Down Expand Up @@ -595,6 +605,7 @@ impl Subscriber {
let repo_root = self.repo_root.clone();
let repository_ignore = self.repository_ignore.clone();
let watch_spec = self.watch_spec.clone();
let watch_spec_ready = self.watch_spec_ready.clone();
let gitignore_path = repo_root.join_component(".gitignore");
let config_paths = [
repo_root.join_component(CONFIG_FILE),
Expand All @@ -620,14 +631,16 @@ impl Subscriber {
let Ok(path) = repo_root.anchor(&absolute_path) else {
return false;
};
if !watch_spec_ready.load(Ordering::Acquire) {
return true;
}
let watch_spec = watch_spec
.read()
.unwrap_or_else(|poisoned| poisoned.into_inner());
let in_ignored_prefix = watch_spec.ignore_prefixes.iter().any(|prefix| {
path.components()
.next()
.is_some_and(|component| component.as_str() == prefix)
});
let in_ignored_prefix = watch_spec
.ignore_prefixes
.iter()
.any(|prefix| path_is_under_prefix(&path, prefix));
if in_ignored_prefix {
return false;
}
Expand Down Expand Up @@ -1001,6 +1014,14 @@ mod test {
.await
}

fn cargo_watch_spec() -> WatchSpec {
WatchSpec {
definition_file_names: vec!["Cargo.toml".to_string()],
definition_paths: vec!["Cargo.lock".to_string()],
ignore_prefixes: vec!["target".to_string()],
}
}

#[tokio::test(flavor = "multi_thread")]
async fn initializes_pure_cargo_and_preserves_root_and_aggregate_scopes() {
let tmp = tempfile::tempdir().unwrap();
Expand All @@ -1016,10 +1037,7 @@ mod test {
.expect("native toolchain permits an absent root package.json");
assert!(!state.pkg_dep_graph.has_root_javascript_scope());
assert!(state.root_turbo_json.is_some());
assert_eq!(
state.pkg_dep_graph.active_watch_spec(),
turborepo_repository::cargo::watch_spec()
);
assert_eq!(state.pkg_dep_graph.active_watch_spec(), cargo_watch_spec());

let scopes: Vec<_> = hash_scopes(&state.pkg_dep_graph).collect();
assert_eq!(scopes.first().unwrap().name, PackageName::Root);
Expand Down Expand Up @@ -1418,8 +1436,7 @@ mod test {

// Manifests define the crate set and its edges; the watcher's graph
// is stale after any manifest change.
let action =
f.classify_with_spec(&trie, &[], None, turborepo_repository::cargo::watch_spec());
let action = f.classify_with_spec(&trie, &[], None, cargo_watch_spec());
assert!(matches!(action, FileChangeAction::ConfigChanged));

// Without the Cargo toolchain registered, the same file is ordinary
Expand All @@ -1435,8 +1452,7 @@ mod test {
let mut trie = Trie::new();
trie.insert(lock.to_string(), ());

let action =
f.classify_with_spec(&trie, &[], None, turborepo_repository::cargo::watch_spec());
let action = f.classify_with_spec(&trie, &[], None, cargo_watch_spec());
assert!(matches!(action, FileChangeAction::ConfigChanged));
}

Expand All @@ -1459,8 +1475,7 @@ mod test {
(),
);

let action =
f.classify_with_spec(&trie, &[], None, turborepo_repository::cargo::watch_spec());
let action = f.classify_with_spec(&trie, &[], None, cargo_watch_spec());
assert!(
matches!(action, FileChangeAction::NoRelevantChanges),
"target/ writes must be dropped, got {action:?}"
Expand All @@ -1472,6 +1487,24 @@ mod test {
assert!(matches!(action, FileChangeAction::PackagesChanged(..)));
}

#[tokio::test]
async fn classify_nested_target_directory_prefix_writes_ignored() {
let f = ClassifyFixture::new().await;
let mut trie = Trie::new();
trie.insert(
f.repo_root
.join_components(&["build", "cargo", "debug", "app"])
.to_string(),
(),
);
let spec = WatchSpec {
ignore_prefixes: vec!["build/cargo".to_string()],
..WatchSpec::default()
};
let action = f.classify_with_spec(&trie, &[], None, spec);
assert!(matches!(action, FileChangeAction::NoRelevantChanges));
}

#[tokio::test]
async fn classify_custom_turbo_json_triggers_config_changed() {
let f = ClassifyFixture::new().await;
Expand Down
Loading
Loading