use std::collections::BTreeMap;
use std::collections::HashSet;
use std::io::ErrorKind;
use std::path::Path;
use std::path::PathBuf;
use std::sync::Arc;
use std::sync::Mutex;
use std::thread;
use cargo_metadata::Error;
use cargo_metadata::Metadata;
use tokio::runtime::Handle;
use tokio::sync::Semaphore;
use walkdir::WalkDir;
use super::BackgroundMsg;
use super::constants::CARGO_OFFLINE_FLAG;
use super::discovery;
use super::disk_usage;
use super::language_stats;
use super::test_counts;
use super::tree;
use crate::channel;
use crate::channel::Receiver;
use crate::channel::Sender;
use crate::config::NonRustInclusion;
use crate::constants::CARGO_METADATA_TIMEOUT;
use crate::constants::CARGO_TOML;
use crate::constants::SCAN_DISK_CONCURRENCY;
use crate::constants::SCAN_METADATA_CONCURRENCY;
use crate::http::HttpClient;
use crate::project::AbsolutePath;
use crate::project::FileStamp;
use crate::project::ManifestFingerprint;
use crate::project::PackageRecord;
use crate::project::PublishPolicy;
use crate::project::RootItem;
use crate::project::TargetRecord;
use crate::project::WorkspaceMetadata;
use crate::project::WorkspaceMetadataStore;
#[derive(Clone, Debug)]
pub(crate) enum CargoMetadataError {
WorkspaceMissing,
Other(String),
}
impl CargoMetadataError {
pub(crate) const fn user_facing_message(&self) -> Option<&str> {
match self {
Self::WorkspaceMissing => None,
Self::Other(message) => Some(message.as_str()),
}
}
}
#[derive(Clone)]
pub(super) struct StreamingScanContext {
pub(super) client: HttpClient,
pub(super) tx: Sender<BackgroundMsg>,
pub(super) disk_limit: Arc<Semaphore>,
pub(super) metadata_store: Arc<Mutex<WorkspaceMetadataStore>>,
pub(super) metadata_limit: Arc<Semaphore>,
}
pub(crate) fn spawn_streaming_scan(
scan_dirs: Vec<AbsolutePath>,
inline_dirs: &[String],
non_rust: NonRustInclusion,
client: HttpClient,
metadata_store: Arc<Mutex<WorkspaceMetadataStore>>,
) -> (Sender<BackgroundMsg>, Receiver<BackgroundMsg>) {
let (tx, rx) = channel::unbounded();
let inline_dirs = inline_dirs.to_vec();
let scan_tx = tx.clone();
thread::spawn(move || {
let scan_context = StreamingScanContext {
client,
tx: scan_tx.clone(),
disk_limit: Arc::new(tokio::sync::Semaphore::new(SCAN_DISK_CONCURRENCY)),
metadata_store,
metadata_limit: Arc::new(tokio::sync::Semaphore::new(SCAN_METADATA_CONCURRENCY)),
};
let phase1_started = std::time::Instant::now();
let phase1 = discovery::phase1_discover(&scan_dirs, non_rust);
tracing::trace!(
target: tui_pane::PERF_LOG_TARGET,
elapsed_ms = tui_pane::perf_log_ms(phase1_started.elapsed().as_millis()),
scan_dirs = scan_dirs.len(),
visited_dirs = phase1.stats.visited_dirs,
manifests = phase1.stats.manifests,
projects = phase1.stats.projects,
non_rust_projects = phase1.stats.non_rust_projects,
disk_entries = phase1.disk_entries.len(),
"phase1_discover_total"
);
let tree_started = std::time::Instant::now();
let projects = tree::build_tree(&phase1.items, &inline_dirs);
tracing::trace!(
target: tui_pane::PERF_LOG_TARGET,
elapsed_ms = tui_pane::perf_log_ms(tree_started.elapsed().as_millis()),
input_items = phase1.items.len(),
tree_items = projects.len(),
"scan_tree_build"
);
let workspace_roots = collect_cargo_metadata_roots(&projects);
let _ = scan_tx.send(BackgroundMsg::ScanResult {
projects,
disk_entries: phase1.disk_entries.clone(),
});
disk_usage::spawn_initial_disk_usage(&scan_context, &phase1.disk_entries);
language_stats::spawn_initial_language_stats(&scan_context, &phase1.disk_entries);
test_counts::spawn_initial_test_counts(&scan_context, &phase1.disk_entries);
spawn_cargo_metadata_tree(&scan_context, workspace_roots);
});
(tx, rx)
}
fn collect_cargo_metadata_roots(projects: &[RootItem]) -> Vec<AbsolutePath> {
let mut seen: HashSet<AbsolutePath> = HashSet::new();
let mut roots = Vec::new();
for item in projects {
for root in cargo_metadata_roots_for_item(item) {
if seen.insert(root.clone()) {
roots.push(root);
}
}
}
roots
}
pub(crate) fn cargo_metadata_roots_for_item(item: &RootItem) -> Vec<AbsolutePath> {
match item {
RootItem::Rust(rust) => vec![rust.path().clone()],
RootItem::Worktrees(group) => group.iter_paths().cloned().collect(),
RootItem::NonRust(_) => Vec::new(),
}
}
fn spawn_cargo_metadata_tree(scan_context: &StreamingScanContext, roots: Vec<AbsolutePath>) {
for workspace_root in roots {
let dispatch = MetadataDispatchContext {
handle: scan_context.client.handle.clone(),
tx: scan_context.tx.clone(),
metadata_store: Arc::clone(&scan_context.metadata_store),
metadata_limit: Arc::clone(&scan_context.metadata_limit),
};
spawn_cargo_metadata_refresh(dispatch, workspace_root);
}
}
#[derive(Clone)]
pub(crate) struct MetadataDispatchContext {
pub handle: Handle,
pub tx: Sender<BackgroundMsg>,
pub metadata_store: Arc<Mutex<WorkspaceMetadataStore>>,
pub metadata_limit: Arc<Semaphore>,
}
impl MetadataDispatchContext {
pub(crate) fn resolved_target_dir(&self, path: &AbsolutePath) -> Option<AbsolutePath> {
self.metadata_store
.lock()
.ok()
.and_then(|store| store.resolved_target_dir(path).cloned())
}
}
pub(crate) fn spawn_cargo_metadata_refresh(
dispatch: MetadataDispatchContext,
workspace_root: AbsolutePath,
) {
let MetadataDispatchContext {
handle,
tx,
metadata_store: store,
metadata_limit: limit,
} = dispatch;
handle.spawn(async move {
let Ok(_permit) = limit.acquire_owned().await else {
return;
};
let workspace_root_for_task = workspace_root.clone();
let blocking = tokio::task::spawn_blocking(move || {
run_cargo_metadata_for_root(&workspace_root_for_task, &store)
});
let task_result = match tokio::time::timeout(CARGO_METADATA_TIMEOUT, blocking).await {
Ok(Ok(output)) => output,
Ok(Err(_)) => {
tracing::warn!(
workspace_root = %workspace_root.display(),
"cargo_metadata_task_join_failed"
);
return;
},
Err(_) => {
let fingerprint = ManifestFingerprint::capture(workspace_root.as_path())
.unwrap_or_else(|_| synthetic_fingerprint());
CargoMetadataTaskOutput {
generation: 0,
fingerprint,
result: Err(CargoMetadataError::Other(format!(
"cargo metadata timed out after {}s",
CARGO_METADATA_TIMEOUT.as_secs()
))),
}
},
};
let CargoMetadataTaskOutput {
generation,
fingerprint,
result,
} = task_result;
let _ = tx.send(BackgroundMsg::CargoMetadata {
workspace_root,
generation,
fingerprint,
result,
});
});
}
struct CargoMetadataTaskOutput {
generation: u64,
fingerprint: ManifestFingerprint,
result: Result<WorkspaceMetadata, CargoMetadataError>,
}
pub(crate) fn spawn_out_of_tree_target_walk(
handle: &Handle,
tx: Sender<BackgroundMsg>,
workspace_root: AbsolutePath,
target_dir: AbsolutePath,
) {
handle.spawn(async move {
let walk_target = target_dir.clone();
let bytes = tokio::task::spawn_blocking(move || sum_dir_bytes(walk_target.as_path())).await;
let bytes = match bytes {
Ok(bytes) => bytes,
Err(err) => {
tracing::warn!(
workspace_root = %workspace_root.display(),
target_dir = %target_dir.display(),
error = %err,
"out_of_tree_target_walk_join_failed"
);
return;
},
};
tracing::debug!(
workspace_root = %workspace_root.display(),
target_dir = %target_dir.display(),
bytes,
"out_of_tree_target_walk_done"
);
let _ = tx.send(BackgroundMsg::OutOfTreeTargetSize {
workspace_root,
target_dir,
bytes,
});
});
}
fn sum_dir_bytes(dir: &Path) -> u64 {
WalkDir::new(dir)
.into_iter()
.flatten()
.filter(|entry| entry.file_type().is_file())
.filter_map(|entry| entry.metadata().ok().map(|meta| meta.len()))
.sum()
}
fn run_cargo_metadata_for_root(
workspace_root: &AbsolutePath,
store: &Arc<Mutex<WorkspaceMetadataStore>>,
) -> CargoMetadataTaskOutput {
let generation = store
.lock()
.map_or(0, |mut guard| guard.next_generation(workspace_root));
let fingerprint = match ManifestFingerprint::capture(workspace_root.as_path()) {
Ok(fp) => fp,
Err(err) => {
let result = if err.kind() == ErrorKind::NotFound {
Err(CargoMetadataError::WorkspaceMissing)
} else {
Err(CargoMetadataError::Other(format!(
"fingerprint capture failed: {err}"
)))
};
return CargoMetadataTaskOutput {
generation,
fingerprint: synthetic_fingerprint(),
result,
};
},
};
let manifest_path = workspace_root.as_path().join(CARGO_TOML);
let started_at = std::time::Instant::now();
let result = match execute_cargo_metadata(&manifest_path) {
Ok(metadata) => Ok(build_workspace_metadata(
workspace_root.clone(),
&metadata,
fingerprint.clone(),
)),
Err(err) => Err(err),
};
tracing::trace!(
target: tui_pane::PERF_LOG_TARGET,
elapsed_ms = tui_pane::perf_log_ms(started_at.elapsed().as_millis()),
workspace_root = %workspace_root.display(),
ok = result.is_ok(),
"cargo_metadata_exec"
);
CargoMetadataTaskOutput {
generation,
fingerprint,
result,
}
}
fn execute_cargo_metadata(manifest_path: &Path) -> Result<Metadata, CargoMetadataError> {
let mut cmd = cargo_metadata::MetadataCommand::new();
cmd.manifest_path(manifest_path).no_deps();
cmd.other_options(vec![CARGO_OFFLINE_FLAG.to_string()]);
cmd.exec()
.map_err(|err| CargoMetadataError::Other(format_cargo_metadata_error(&err)))
}
fn format_cargo_metadata_error(err: &Error) -> String {
let text = err.to_string();
text.lines().next().unwrap_or(&text).to_string()
}
const fn synthetic_fingerprint() -> ManifestFingerprint {
ManifestFingerprint {
manifest: FileStamp {
content_hash: [0_u8; 32],
},
lockfile: None,
rust_toolchain: None,
configs: BTreeMap::new(),
}
}
fn build_workspace_metadata(
workspace_root: AbsolutePath,
metadata: &Metadata,
fingerprint: ManifestFingerprint,
) -> WorkspaceMetadata {
let target_directory =
AbsolutePath::from(PathBuf::from(metadata.target_directory.as_std_path()));
let packages = metadata
.packages
.iter()
.map(|pkg| {
let record = PackageRecord {
name: pkg.name.to_string(),
version: pkg.version.clone(),
edition: pkg.edition.to_string(),
description: pkg.description.clone(),
license: pkg.license.clone(),
homepage: pkg.homepage.clone(),
repository: pkg.repository.clone(),
manifest_path: AbsolutePath::from(PathBuf::from(pkg.manifest_path.as_std_path())),
targets: pkg
.targets
.iter()
.map(|target| TargetRecord {
name: target.name.clone(),
kinds: target.kind.clone(),
src_path: AbsolutePath::from(PathBuf::from(
target.src_path.as_std_path(),
)),
required_features: target.required_features.clone(),
})
.collect(),
publish: PublishPolicy::from_cargo_publish(pkg.publish.as_deref()),
};
(pkg.id.clone(), record)
})
.collect();
WorkspaceMetadata {
workspace_root,
target_directory,
packages,
fingerprint,
out_of_tree_target_bytes: None,
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::project::Package;
use crate::project::RustProject;
use crate::project::Workspace;
use crate::project::WorktreeStatus;
use crate::scan::tree;
fn status_for(is_linked_worktree: bool, primary_abs: Option<&str>) -> WorktreeStatus {
match (is_linked_worktree, primary_abs) {
(_, None) => WorktreeStatus::NotGit,
(true, Some(p)) => WorktreeStatus::Linked {
primary: AbsolutePath::from(p.to_string()),
},
(false, Some(p)) => WorktreeStatus::Primary {
root: AbsolutePath::from(p.to_string()),
},
}
}
fn make_workspace(
name: Option<&str>,
abs_path: &str,
is_linked_worktree: bool,
primary_abs: Option<&str>,
) -> RootItem {
RootItem::Rust(RustProject::Workspace(Workspace {
path: AbsolutePath::from(abs_path),
name: name.map(String::from),
worktree_status: status_for(is_linked_worktree, primary_abs),
..Workspace::default()
}))
}
fn make_package(
name: Option<&str>,
abs_path: &str,
is_linked_worktree: bool,
primary_abs: Option<&str>,
) -> RootItem {
RootItem::Rust(RustProject::Package(Package {
path: AbsolutePath::from(abs_path),
name: name.map(String::from),
worktree_status: status_for(is_linked_worktree, primary_abs),
..Package::default()
}))
}
#[test]
fn collect_cargo_metadata_roots_yields_one_root_per_rust_leaf() {
let ws = make_workspace(Some("ws"), "/ws", false, Some("/ws"));
let pkg = make_package(Some("pkg"), "/pkg", false, Some("/pkg"));
let roots = collect_cargo_metadata_roots(&[ws, pkg]);
assert_eq!(
roots,
vec![AbsolutePath::from("/ws"), AbsolutePath::from("/pkg"),],
"each Rust leaf produces exactly one metadata root, preserving input order"
);
}
#[test]
fn collect_cargo_metadata_roots_skips_non_rust_projects() {
let non_rust = RootItem::NonRust(crate::project::NonRustProject::new(
AbsolutePath::from("/notes"),
Some("notes".into()),
));
let pkg = make_package(Some("pkg"), "/pkg", false, Some("/pkg"));
let roots = collect_cargo_metadata_roots(&[non_rust, pkg]);
assert_eq!(
roots,
vec![AbsolutePath::from("/pkg")],
"non-rust leaves never receive a metadata dispatch"
);
}
#[test]
fn collect_cargo_metadata_roots_unions_primary_and_linked_worktrees() {
let primary = make_workspace(Some("ws"), "/ws", false, Some("/ws"));
let linked_a = make_workspace(Some("ws_feat"), "/ws_feat", true, Some("/ws"));
let linked_b = make_workspace(Some("ws_bug"), "/ws_bug", true, Some("/ws"));
let mut items = vec![primary, linked_a, linked_b];
tree::merge_worktrees_new(&mut items);
assert_eq!(items.len(), 1, "merged into one worktree group");
let mut roots = collect_cargo_metadata_roots(&items);
roots.sort_by(|a, b| a.as_path().cmp(b.as_path()));
assert_eq!(
roots,
vec![
AbsolutePath::from("/ws"),
AbsolutePath::from("/ws_bug"),
AbsolutePath::from("/ws_feat"),
],
"primary + every linked worktree gets its own metadata root"
);
}
#[test]
fn collect_cargo_metadata_roots_dedupes_repeated_paths() {
let pkg_a = make_package(Some("a"), "/pkg", false, Some("/pkg"));
let pkg_b = make_package(Some("b"), "/pkg", false, Some("/pkg"));
let roots = collect_cargo_metadata_roots(&[pkg_a, pkg_b]);
assert_eq!(roots, vec![AbsolutePath::from("/pkg")]);
}
}