use std::collections::BTreeMap;
use std::fs::{self, File, OpenOptions};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::{SystemTime, UNIX_EPOCH};
use fs2::FileExt;
use crate::activation::{ActivationProvider, PreparedActivation};
use crate::artifact::{ArtifactProvenance, ArtifactStore, VerifiedArtifact, digest_tree};
use crate::backends::cargo::{ResolvedCargo, SourceMode};
use crate::backends::typed::{
BackendContext, BackendOperation, CommandInvocation, install_operations,
};
use crate::cancellation::{delay as cancellation_delay, requested};
use crate::config::schema::{CargoInstall, InstallSpec, NetworkPolicy};
use crate::error::ForgeError;
use crate::events::{LifecycleEvent, emit};
use crate::execution::command::{
CommandMetrics, ShellDisplayMode, ShellStep, run_invocation_labeled_measured,
};
use crate::execution::install::verify_typed_tool;
use crate::execution::retry::RetryPolicy;
use crate::execution::scheduler::{CancellationToken, ResourceCoordinator, ResourceLeases};
use crate::fsutil::{copy_file, create_dir_all, lock_exclusive_while, remove_dir_all};
use crate::model::{
BackendKind, InstallKind, RegistryEntry, RegistryTarget, ToolDef, ToolInstallOutcome,
ToolInstallResult,
};
use crate::paths::{app_home, managed_bin_dir};
use crate::planning::ExecutionPlan;
use crate::planning::ResourceClaim;
use crate::state::journal::{JournalCheckpoint, JournalPhase};
use crate::state::{append_registry, read_registry};
use crate::telemetry::{estimates, record_cargo, record_corruption};
use crate::util::{exe_name, fnv1a, home_dir, now_secs};
pub(crate) fn install(
plan: &ExecutionPlan,
tool: &ToolDef,
specification: &CargoInstall,
display_mode: ShellDisplayMode,
step: Option<ShellStep>,
) -> Result<ToolInstallResult, ForgeError> {
install_with_jobs(plan, tool, specification, display_mode, step, 1)
}
pub(crate) fn install_with_jobs(
plan: &ExecutionPlan,
tool: &ToolDef,
specification: &CargoInstall,
display_mode: ShellDisplayMode,
step: Option<ShellStep>,
cargo_jobs: usize,
) -> Result<ToolInstallResult, ForgeError> {
install_transaction(
plan,
tool,
specification,
display_mode,
step,
cargo_jobs,
None,
)
}
pub(crate) fn install_coordinated(
plan: &ExecutionPlan,
tool: &ToolDef,
specification: &CargoInstall,
display_mode: ShellDisplayMode,
cargo_jobs: usize,
resources: CoordinatedCargoResources<'_>,
) -> Result<ToolInstallResult, ForgeError> {
install_transaction(
plan,
tool,
specification,
display_mode,
None,
cargo_jobs,
Some(resources),
)
}
#[derive(Clone, Copy)]
pub(crate) struct CoordinatedCargoResources<'a> {
pub(crate) coordinator: &'a ResourceCoordinator,
pub(crate) remaining_builds: &'a AtomicUsize,
pub(crate) baseline_jobs: usize,
pub(crate) max_builds: usize,
pub(crate) cancellation: &'a CancellationToken,
}
fn install_transaction(
plan: &ExecutionPlan,
tool: &ToolDef,
specification: &CargoInstall,
display_mode: ShellDisplayMode,
step: Option<ShellStep>,
cargo_jobs: usize,
resources: Option<CoordinatedCargoResources<'_>>,
) -> Result<ToolInstallResult, ForgeError> {
let _remaining = resources.map(|resources| RemainingBuildGuard(resources.remaining_builds));
let work_root = app_home().join("work").join("cargo").join(&tool.name);
let source_mode = match plan.policy.network {
NetworkPolicy::Online => SourceMode::Online,
NetworkPolicy::CacheOnly => SourceMode::CacheOnly,
NetworkPolicy::Offline => SourceMode::Offline,
};
let resolved = ResolvedCargo::resolve(&tool.name, specification, &work_root, source_mode)?;
let cancellation = resources.map(|resources| resources.cancellation);
let _work_lease = CargoWorkLease::acquire(&resolved.fingerprint, cancellation)?;
let cache_digest = specification.cache_digest();
let source_home = app_home()
.join("cache")
.join("cargo")
.join("sources")
.join(cache_digest);
let previous_artifact_id = read_registry()?
.into_iter()
.find(|entry| entry.name == tool.name && entry.kind == InstallKind::Tool)
.and_then(|entry| entry.artifact_id);
let mut journal = JournalCheckpoint::new(
run_id(&plan.plan_hash),
&tool.name,
&plan.plan_hash,
&plan.config_hash,
&plan.profile,
);
journal.save()?;
let provenance = cargo_provenance(&resolved, specification);
let mut memory_mib = estimates()
.get(&resolved.fingerprint)
.map_or(512, |estimate| estimate.memory_mib);
if let Some(resources) = resources {
memory_mib = memory_mib.min(resources.coordinator.capacity("memory-mib").max(512));
}
let artifact_id = format!("{}-{}", resolved.component, resolved.fingerprint);
let store = ArtifactStore::new(None)?;
let lookup = {
let _lease = acquire_stage(
resources,
vec![claim(&format!("artifact:{}", resolved.fingerprint), 1)],
&tool.name,
)?;
store.get(&artifact_id, &provenance, &specification.bins)?
};
if lookup.quarantined {
record_corruption(&resolved.fingerprint, &tool.name);
}
let cached = lookup.artifact;
let cache_hit = cached.is_some();
let mut actual_cargo_jobs = cargo_jobs.max(1);
let mut build_ms = 0;
let mut peak_rss_mib = 0;
let (artifact, store_root) = if let Some(artifact) = cached {
let root = artifact.root.clone();
(artifact, root)
} else {
create_dir_all(&source_home)?;
reset_work_directory(&resolved.staging)?;
let _staging_cleanup = DirectoryCleanup::new(resolved.staging.clone());
create_dir_all(&resolved.target_dir)?;
{
let started = std::time::Instant::now();
let (build_lease, build_jobs) = acquire_build_stage(
resources,
vec![
claim("network", 1),
claim("memory-mib", memory_mib),
claim("disk-io", 1),
claim(&format!("cargo-build:{}", resolved.fingerprint), 1),
],
&tool.name,
cargo_jobs,
)?;
let _lease = build_lease;
actual_cargo_jobs = build_jobs;
peak_rss_mib = run_cargo_install(
&resolved,
specification,
&tool.name,
display_mode,
step,
build_jobs,
&source_home,
)?
.peak_rss_mib;
build_ms = started.elapsed().as_millis().min(u128::from(u64::MAX)) as u64;
}
let artifact = {
let _lease = acquire_stage(resources, vec![claim("disk-io", 1)], &tool.name)?;
collect_artifact(&resolved, specification)?
};
let root = {
let _lease = acquire_stage(
resources,
vec![
claim("disk-io", 1),
claim(&format!("artifact:{}", resolved.fingerprint), 1),
],
&tool.name,
)?;
store.put(&artifact)?
};
if resolved.staging.exists() {
remove_dir_all(&resolved.staging)?;
}
(artifact, root)
};
let stored_binaries = stored_binaries(&artifact, &store_root)?;
let activation_claims = stored_binaries
.keys()
.map(|name| claim(&format!("bin:{name}"), 1))
.collect::<Vec<_>>();
let _activation_lease = acquire_stage(resources, activation_claims, &tool.name)?;
let provider = ActivationProvider::new(managed_bin_dir())?;
let prepared = prepare_activations(&provider, &stored_binaries)?;
let activation_paths = prepared
.iter()
.map(|activation| activation.destination().to_path_buf())
.collect::<Vec<_>>();
journal.update_artifacts(
&artifact.id,
previous_artifact_id.clone(),
activation_paths.clone(),
)?;
journal.transition(JournalPhase::Stored)?;
journal.transition(JournalPhase::Activating)?;
let switched = switch_activations(&prepared, &mut journal)?;
journal.transition(JournalPhase::Activated)?;
if let Err(error) = verify_typed_tool(tool) {
rollback_activations(&prepared, switched)?;
journal.transition(JournalPhase::RolledBack)?;
return Err(ForgeError::Command(format!(
"verification failed for tool {}: {error}",
tool.name
)));
}
let entry = registry_entry(
plan,
tool,
specification,
&artifact,
previous_artifact_id,
activation_paths,
&stored_binaries,
)?;
if let Err(error) = append_registry(entry) {
rollback_activations(&prepared, switched)?;
journal.transition(JournalPhase::RolledBack)?;
return Err(error);
}
journal.transition(JournalPhase::Recorded)?;
record_cargo(
&resolved.fingerprint,
&tool.name,
build_ms,
cache_hit,
actual_cargo_jobs,
peak_rss_mib,
);
let version = specification.version.trim_start_matches('=');
Ok(ToolInstallResult {
name: tool.name.clone(),
outcome: ToolInstallOutcome::Installed,
message: Some(version.to_string()),
})
}
struct RemainingBuildGuard<'a>(&'a AtomicUsize);
impl Drop for RemainingBuildGuard<'_> {
fn drop(&mut self) {
self.0.fetch_sub(1, Ordering::AcqRel);
}
}
fn claim(key: &str, units: u32) -> ResourceClaim {
ResourceClaim {
key: key.to_string(),
units,
}
}
fn acquire_stage<'a>(
resources: Option<CoordinatedCargoResources<'a>>,
claims: Vec<ResourceClaim>,
component: &str,
) -> Result<Option<ResourceLeases<'a>>, ForgeError> {
let Some(resources) = resources else {
return Ok(None);
};
let coordinator = resources.coordinator;
let cancellation = resources.cancellation;
let resource = claims
.iter()
.map(|claim| claim.key.as_str())
.collect::<Vec<_>>()
.join(",");
emit(
None,
Some(component),
Some("cargo"),
LifecycleEvent::ResourceWait { resource },
);
let started = std::time::Instant::now();
let leases = coordinator.acquire(&claims, cancellation)?;
emit(
None,
Some(component),
Some("cargo"),
LifecycleEvent::ResourceAcquired {
resource: claims
.iter()
.map(|claim| claim.key.as_str())
.collect::<Vec<_>>()
.join(","),
wait_ms: started.elapsed().as_millis(),
},
);
Ok(Some(leases))
}
fn acquire_build_stage<'a>(
resources: Option<CoordinatedCargoResources<'a>>,
claims: Vec<ResourceClaim>,
component: &str,
fallback_jobs: usize,
) -> Result<(Option<ResourceLeases<'a>>, usize), ForgeError> {
let Some(resources) = resources else {
return Ok((None, fallback_jobs.max(1)));
};
let coordinator = resources.coordinator;
let cancellation = resources.cancellation;
let remaining = resources.remaining_builds.load(Ordering::Acquire).max(1);
emit(
None,
Some(component),
Some("cargo"),
LifecycleEvent::ResourceWait {
resource: "dynamic-cargo-build".into(),
},
);
let started = std::time::Instant::now();
let (leases, jobs) = coordinator.acquire_cargo_build(
claims,
remaining,
resources.baseline_jobs,
resources.max_builds,
cancellation,
)?;
emit(
None,
Some(component),
Some("cargo"),
LifecycleEvent::ResourceAcquired {
resource: format!("dynamic-cargo-build:{jobs}-jobs"),
wait_ms: started.elapsed().as_millis(),
},
);
Ok((Some(leases), jobs))
}
struct CargoWorkLease {
file: File,
}
impl CargoWorkLease {
fn acquire(
fingerprint: &str,
cancellation: Option<&CancellationToken>,
) -> Result<Self, ForgeError> {
let directory = app_home().join("locks").join("cargo-work");
create_dir_all(&directory)?;
let path = directory.join(format!("{fingerprint}.lock"));
let file = OpenOptions::new()
.create(true)
.truncate(false)
.read(true)
.write(true)
.open(&path)
.map_err(|source| ForgeError::Io {
path: path.clone(),
source,
})?;
lock_exclusive_while(&file, &path, "Cargo work lock", || {
cancellation.is_some_and(CancellationToken::is_cancelled) || requested()
})?;
Ok(Self { file })
}
}
impl Drop for CargoWorkLease {
fn drop(&mut self) {
let _ = FileExt::unlock(&self.file);
}
}
fn reset_work_directory(path: &Path) -> Result<(), ForgeError> {
if path.exists() {
remove_dir_all(path)?;
}
create_dir_all(path)
}
struct DirectoryCleanup {
path: PathBuf,
}
impl DirectoryCleanup {
fn new(path: PathBuf) -> Self {
Self { path }
}
}
impl Drop for DirectoryCleanup {
fn drop(&mut self) {
if self.path.exists() {
let _ = remove_dir_all(&self.path);
}
}
}
fn run_cargo_install(
resolved: &ResolvedCargo,
specification: &CargoInstall,
label: &str,
display_mode: ShellDisplayMode,
step: Option<ShellStep>,
cargo_jobs: usize,
source_home: &Path,
) -> Result<CommandMetrics, ForgeError> {
let mut metrics = CommandMetrics::default();
let context = BackendContext {
staging: &resolved.staging,
target: &resolved.target_dir,
cargo_jobs: cargo_jobs.max(1),
allow_insecure_hosts: &[],
apt_source: None,
apt_lists: None,
};
for operation in install_operations(&InstallSpec::Cargo(specification.clone()), &context)? {
let BackendOperation::Command(mut invocation) = operation else {
return Err(ForgeError::Config(
"Cargo backend produced a non-command operation".into(),
));
};
if matches!(
resolved.source_mode,
SourceMode::CacheOnly | SourceMode::Offline
) {
invocation.args.push("--offline".into());
}
let user_home = std::env::var_os("CARGO_HOME")
.filter(|value| !value.is_empty())
.map(PathBuf::from)
.unwrap_or_else(|| home_dir().join(".cargo"));
if let Some(config) = [user_home.join("config"), user_home.join("config.toml")]
.into_iter()
.find(|path| path.is_file())
{
let absolute = std::path::absolute(&config).map_err(|source| ForgeError::Io {
path: config,
source,
})?;
invocation
.args
.extend(["--config".into(), absolute.display().to_string()]);
}
invocation
.env
.insert("CARGO_HOME".into(), source_home.display().to_string());
let measured = run_cargo_invocation_with_retry(label, &invocation, display_mode, step)?;
metrics.peak_rss_mib = metrics.peak_rss_mib.max(measured.peak_rss_mib);
}
Ok(metrics)
}
fn run_cargo_invocation_with_retry(
label: &str,
invocation: &CommandInvocation,
display_mode: ShellDisplayMode,
step: Option<ShellStep>,
) -> Result<CommandMetrics, ForgeError> {
let policy = RetryPolicy {
max_attempts: 3,
base_delay: std::time::Duration::from_millis(200),
max_delay: std::time::Duration::from_secs(2),
};
let mut attempt = 0;
loop {
match run_invocation_labeled_measured(label, invocation, display_mode, step) {
Ok(metrics) => return Ok(metrics),
Err(error) if policy.should_retry(attempt, invocation.idempotent, &error) => {
let delay = policy.delay(attempt, fnv1a(label));
emit(
None,
Some(label),
Some("cargo"),
LifecycleEvent::RetryScheduled {
attempt: attempt + 2,
delay_ms: delay.as_millis().min(u128::from(u64::MAX)) as u64,
},
);
cancellation_delay(delay, "Cargo retry backoff")?;
attempt += 1;
}
Err(error) => return Err(error),
}
}
}
fn collect_artifact(
resolved: &ResolvedCargo,
specification: &CargoInstall,
) -> Result<VerifiedArtifact, ForgeError> {
let installed_bin = resolved.staging.join("bin");
if !installed_bin.is_dir() {
return Err(ForgeError::Config(format!(
"Cargo install did not produce a bin directory: {}",
installed_bin.display()
)));
}
let names = installed_binary_names(&installed_bin, &specification.bins)?;
let artifact_root = resolved.staging.join("verified");
create_dir_all(&artifact_root)?;
let mut binaries = BTreeMap::new();
for name in names {
let source = installed_bin.join(exe_name(&name));
let destination = artifact_root.join(exe_name(&name));
if !source.is_file() {
return Err(ForgeError::Config(format!(
"Cargo binary does not exist: {}",
source.display()
)));
}
copy_file(&source, &destination)?;
binaries.insert(name, destination);
}
let sha256 = digest_tree(&artifact_root)?;
Ok(VerifiedArtifact {
id: format!("{}-{}", resolved.component, resolved.fingerprint),
root: artifact_root,
binaries,
sha256,
provenance: cargo_provenance(resolved, specification),
})
}
fn cargo_provenance(resolved: &ResolvedCargo, specification: &CargoInstall) -> ArtifactProvenance {
ArtifactProvenance {
backend: "cargo".into(),
source: specification.crate_name.clone(),
revision: specification.revision.clone(),
fingerprint: resolved.fingerprint.clone(),
}
}
fn installed_binary_names(bin: &Path, expected: &[String]) -> Result<Vec<String>, ForgeError> {
let mut actual = fs::read_dir(bin)
.map_err(|source| ForgeError::Io {
path: bin.to_path_buf(),
source,
})?
.filter_map(Result::ok)
.filter(|entry| entry.path().is_file())
.filter_map(|entry| binary_stem(&entry.path()))
.collect::<Vec<_>>();
actual.sort();
actual.dedup();
if actual.is_empty() {
return Err(ForgeError::Config(
"Cargo install did not produce a binary".into(),
));
}
if !expected.is_empty() {
let mut expected = expected.to_vec();
expected.sort();
expected.dedup();
if actual != expected {
return Err(ForgeError::Config(format!(
"Cargo binary set does not match; expected {}, found {}",
expected.join(", "),
actual.join(", ")
)));
}
}
Ok(actual)
}
fn binary_stem(path: &Path) -> Option<String> {
let name = path.file_name()?.to_str()?;
if cfg!(windows) {
name.strip_suffix(".exe").map(str::to_string)
} else {
Some(name.to_string())
}
}
fn stored_binaries(
artifact: &VerifiedArtifact,
store_root: &Path,
) -> Result<BTreeMap<String, PathBuf>, ForgeError> {
artifact
.binaries
.iter()
.map(|(name, path)| {
let relative = path.strip_prefix(&artifact.root).map_err(|_| {
ForgeError::Config(format!(
"artifact binary escapes the verified root: {}",
path.display()
))
})?;
Ok((name.clone(), store_root.join(relative)))
})
.collect()
}
fn prepare_activations(
provider: &ActivationProvider,
binaries: &BTreeMap<String, PathBuf>,
) -> Result<Vec<PreparedActivation>, ForgeError> {
binaries
.iter()
.map(|(name, executable)| provider.prepare(name, executable))
.collect()
}
pub(crate) fn switch_activations(
prepared: &[PreparedActivation],
journal: &mut JournalCheckpoint,
) -> Result<usize, ForgeError> {
for (index, activation) in prepared.iter().enumerate() {
if let Err(error) = activation.switch() {
rollback_activations(prepared, index)?;
journal.transition(JournalPhase::RolledBack)?;
return Err(error);
}
}
Ok(prepared.len())
}
pub(crate) fn rollback_activations(
prepared: &[PreparedActivation],
switched: usize,
) -> Result<(), ForgeError> {
let mut failure = None;
for activation in prepared[..switched].iter().rev() {
if let Err(error) = activation.restore() {
failure.get_or_insert(error);
}
}
failure.map_or(Ok(()), Err)
}
fn registry_entry(
plan: &ExecutionPlan,
tool: &ToolDef,
specification: &CargoInstall,
artifact: &VerifiedArtifact,
previous_artifact_id: Option<String>,
targets: Vec<PathBuf>,
stored_binaries: &BTreeMap<String, PathBuf>,
) -> Result<RegistryEntry, ForgeError> {
let targets = stored_binaries
.keys()
.cloned()
.zip(targets)
.map(|(binary, path)| RegistryTarget {
path,
binary: Some(binary),
})
.collect();
Ok(RegistryEntry {
name: tool.name.clone(),
kind: InstallKind::Tool,
source: "backend:cargo".into(),
profile: plan.profile.clone(),
targets,
installed_at: now_secs(),
artifact_id: Some(artifact.id.clone()),
previous_artifact_id,
config_hash: Some(plan.config_hash.clone()),
plan_hash: Some(plan.plan_hash.clone()),
source_revision: specification.revision.clone(),
backend: Some(BackendKind::Cargo),
})
}
fn run_id(plan_hash: &str) -> String {
let nanos = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map_or(0, |duration| duration.as_nanos());
format!(
"{}-{}-{nanos}",
&plan_hash[..plan_hash.len().min(12)],
std::process::id()
)
}
#[cfg(test)]
mod tests {
use std::fs;
use crate::execution::managed_cargo::{DirectoryCleanup, installed_binary_names};
use crate::util::{exe_name, now_secs};
#[test]
fn validates_complete_multi_binary_set() {
let root = std::env::temp_dir().join(format!(
"bot-forge-managed-cargo-{}-{}",
std::process::id(),
now_secs()
));
fs::create_dir_all(&root).unwrap();
fs::write(root.join(exe_name("alpha")), "a").unwrap();
fs::write(root.join(exe_name("beta")), "b").unwrap();
assert_eq!(
installed_binary_names(&root, &["beta".into(), "alpha".into()]).unwrap(),
["alpha", "beta"]
);
assert!(installed_binary_names(&root, &["alpha".into()]).is_err());
fs::remove_dir_all(root).unwrap();
}
#[test]
fn failed_transaction_cleanup_removes_staging() {
let root = std::env::temp_dir().join(format!(
"bot-forge-cargo-cleanup-{}-{}",
std::process::id(),
now_secs()
));
fs::create_dir_all(root.join("partial")).unwrap();
fs::write(root.join("partial/output"), "partial").unwrap();
{
let _cleanup = DirectoryCleanup::new(root.clone());
}
assert!(!root.exists());
}
}