use super::ssh::{run_offload_ssh_command_with_stdin, should_skip_remote_preflight};
use super::*;
pub(super) fn merge_sync_result(base: &SyncResult, extra: &SyncResult) -> SyncResult {
SyncResult {
bytes_transferred: base
.bytes_transferred
.saturating_add(extra.bytes_transferred),
files_transferred: base
.files_transferred
.saturating_add(extra.files_transferred),
duration_ms: base.duration_ms.saturating_add(extra.duration_ms),
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub(super) struct SyncClosurePlanEntry {
pub(super) local_root: PathBuf,
pub(super) remote_root: String,
pub(super) project_id: String,
pub(super) root_hash: String,
pub(super) is_primary: bool,
pub(super) mode: SyncClosureMode,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize)]
#[serde(rename_all = "snake_case")]
pub(super) enum SyncClosureMode {
Full,
WorkspaceMetadata,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub(super) struct OwnedSyncClosureLayout {
pub(super) source_root: String,
pub(super) pooled_target_name: String,
pub(super) plan: Vec<SyncClosurePlanEntry>,
}
#[derive(Debug, thiserror::Error)]
pub(super) enum OwnedSyncClosureLayoutError {
#[error("invalid owned closure staging base: {0}")]
InvalidBase(String),
#[error("owned closure requires a valid pooled Cargo target name")]
InvalidPoolName,
#[error("invalid owned closure plan: {0}")]
InvalidPlan(&'static str),
#[error("invalid owned closure root {root}: {reason}")]
InvalidRoot { root: PathBuf, reason: String },
#[error("owned closure roots have overlapping transfer selections: {first} and {second}")]
OverlappingRoots { first: PathBuf, second: PathBuf },
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub(super) enum SyncRootOutcome {
Synced,
Skipped { reason: String },
Failed { error: String },
}
#[derive(Debug, Clone, Serialize, PartialEq, Eq)]
pub(super) struct SyncClosureManifest {
pub(super) schema_version: &'static str,
pub(super) generated_at_unix_ms: i64,
pub(super) project_root: String,
pub(super) entries: Vec<SyncClosureManifestEntry>,
}
#[derive(Debug, Clone, Serialize, PartialEq, Eq)]
pub(super) struct SyncClosureManifestEntry {
pub(super) order: usize,
pub(super) local_root: String,
pub(super) remote_root: String,
pub(super) project_id: String,
pub(super) root_hash: String,
pub(super) is_primary: bool,
mode: SyncClosureMode,
}
pub(super) const DEPENDENCY_PREFLIGHT_SCHEMA_VERSION: &str = "rch.dependency_preflight.v1";
const DEPENDENCY_PREFLIGHT_CODE_PRESENT: &str = "RCH-I410";
pub(super) const DEPENDENCY_PREFLIGHT_CODE_MISSING: &str = "RCH-E410";
pub(super) const DEPENDENCY_PREFLIGHT_CODE_STALE: &str = "RCH-E411";
pub(super) const DEPENDENCY_PREFLIGHT_CODE_UNKNOWN: &str = "RCH-E412";
pub(super) const DEPENDENCY_PREFLIGHT_CODE_POLICY: &str = "RCH-E413";
pub(super) const DEPENDENCY_PREFLIGHT_CODE_TIMEOUT: &str = "RCH-E414";
pub(super) const DEPENDENCY_PREFLIGHT_CODE_MATERIALIZATION: &str = "RCH-E415";
pub(super) const DEPENDENCY_PREFLIGHT_REMEDIATION_MISSING: &str = "Ensure every dependency root in the closure is synced and Cargo.toml plus required source entrypoints exist remotely.";
pub(super) const DEPENDENCY_PREFLIGHT_REMEDIATION_STALE: &str = "One or more dependency roots were not refreshed; rerun after successful sync of skipped roots.";
pub(super) const DEPENDENCY_PREFLIGHT_REMEDIATION_UNKNOWN: &str =
"Dependency verification could not determine remote state; inspect sync/SSH logs and retry.";
pub(super) const DEPENDENCY_PREFLIGHT_REMEDIATION_POLICY: &str = "Path dependency topology policy failed; move dependencies under the canonical project root (path_topology.canonical_root, default /data/projects, alias /dp) and retry.";
pub(super) const DEPENDENCY_PREFLIGHT_REMEDIATION_TIMEOUT: &str = "Dependency planner timed out; rerun after system load decreases or investigate cargo metadata latency.";
pub(super) const DEPENDENCY_PREFLIGHT_REMEDIATION_MATERIALIZATION: &str = "Cargo path materialization is incomplete; repair the reported repository manifest/path dependency before retrying remote Cargo execution.";
const DEPENDENCY_PREFLIGHT_PATHS_PER_TIMEOUT_WINDOW: usize = 128;
const WORKSPACE_METADATA_SYNC_PATTERNS: &[&str] = &[
"Cargo.toml",
"Cargo.lock",
"rust-toolchain",
"rust-toolchain.toml",
".cargo/",
".cargo/**",
];
#[derive(Debug, Clone, Copy, Serialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub(super) enum DependencyPreflightStatus {
Present,
Missing,
Stale,
PolicyViolation,
Timeout,
MaterializationUnavailable,
Unknown,
}
#[derive(Debug, Clone, Serialize, PartialEq, Eq)]
pub(super) struct DependencyPreflightEvidence {
pub(super) root: String,
pub(super) manifest: String,
pub(super) required_path: String,
pub(super) required_kind: &'static str,
pub(super) status: DependencyPreflightStatus,
pub(super) reason_code: &'static str,
pub(super) detail: String,
pub(super) is_primary: bool,
}
#[derive(Debug, Clone, Serialize, PartialEq, Eq)]
pub(super) struct DependencyPreflightReport {
pub(super) schema_version: &'static str,
pub(super) worker: String,
pub(super) verified: bool,
pub(super) reason_code: Option<&'static str>,
pub(super) remediation: Option<&'static str>,
pub(super) evidence: Vec<DependencyPreflightEvidence>,
}
#[derive(Debug, thiserror::Error)]
#[error("dependency preflight verification failed [{reason_code}]")]
pub(super) struct DependencyPreflightFailure {
pub(super) reason_code: &'static str,
pub(super) remediation: String,
report: DependencyPreflightReport,
}
impl DependencyPreflightFailure {
pub(super) fn from_report(report: DependencyPreflightReport) -> Self {
let reason_code = report
.reason_code
.unwrap_or(DEPENDENCY_PREFLIGHT_CODE_UNKNOWN);
let remediation = report
.remediation
.unwrap_or(DEPENDENCY_PREFLIGHT_REMEDIATION_UNKNOWN);
Self {
reason_code,
remediation: dependency_preflight_remediation(reason_code, remediation, &report.worker),
report,
}
}
pub(super) fn report_json(&self) -> String {
serde_json::to_string(&self.report).unwrap_or_else(|err| {
format!(
"{{\"schema_version\":\"{}\",\"verified\":false,\"reason_code\":\"{}\",\"serialization_error\":\"{}\"}}",
DEPENDENCY_PREFLIGHT_SCHEMA_VERSION, self.reason_code, err
)
})
}
pub(super) fn evidence_summary(&self) -> String {
dependency_preflight_evidence_summary(&self.report.evidence)
}
}
fn dependency_preflight_remediation(
reason_code: &str,
remediation: &str,
worker_id: &str,
) -> String {
if matches!(
reason_code,
DEPENDENCY_PREFLIGHT_CODE_MISSING | DEPENDENCY_PREFLIGHT_CODE_STALE
) {
let worker = shell_escape::escape(worker_id.into());
return format!(
"{remediation} To explicitly refresh this worker's RCH-managed source cache, run `rch sync --worker {worker} --force`, then rerun the Cargo command."
);
}
remediation.to_string()
}
fn dependency_preflight_status_label(status: DependencyPreflightStatus) -> &'static str {
match status {
DependencyPreflightStatus::Present => "present",
DependencyPreflightStatus::Missing => "missing",
DependencyPreflightStatus::Stale => "stale",
DependencyPreflightStatus::PolicyViolation => "policy_violation",
DependencyPreflightStatus::Timeout => "timeout",
DependencyPreflightStatus::MaterializationUnavailable => "materialization_unavailable",
DependencyPreflightStatus::Unknown => "unknown",
}
}
fn dependency_preflight_evidence_summary(evidence: &[DependencyPreflightEvidence]) -> String {
let selected = evidence
.iter()
.find(|item| item.status != DependencyPreflightStatus::Present)
.or_else(|| evidence.first());
match selected {
Some(item) => format!(
"{} {} {}: {}",
dependency_preflight_status_label(item.status),
item.required_kind,
item.required_path,
item.detail
),
None => "no dependency preflight evidence rows".to_string(),
}
}
pub(super) fn parse_dependency_preflight_probe_output(
stdout: &str,
) -> (
std::collections::BTreeSet<String>,
std::collections::BTreeSet<String>,
) {
let mut present = std::collections::BTreeSet::new();
let mut missing = std::collections::BTreeSet::new();
for line in stdout.lines() {
if let Some(path) = line.strip_prefix("RCH_DEP_PRESENT:") {
present.insert(path.trim().to_string());
} else if let Some(path) = line.strip_prefix("RCH_DEP_MISSING:") {
missing.insert(path.trim().to_string());
}
}
(present, missing)
}
fn dependency_preflight_failure_reason(
evidence: &[DependencyPreflightEvidence],
) -> Option<(&'static str, &'static str)> {
if evidence
.iter()
.any(|item| item.status == DependencyPreflightStatus::Missing)
{
return Some((
DEPENDENCY_PREFLIGHT_CODE_MISSING,
DEPENDENCY_PREFLIGHT_REMEDIATION_MISSING,
));
}
if evidence
.iter()
.any(|item| item.status == DependencyPreflightStatus::Stale)
{
return Some((
DEPENDENCY_PREFLIGHT_CODE_STALE,
DEPENDENCY_PREFLIGHT_REMEDIATION_STALE,
));
}
if evidence
.iter()
.any(|item| item.status == DependencyPreflightStatus::MaterializationUnavailable)
{
return Some((
DEPENDENCY_PREFLIGHT_CODE_MATERIALIZATION,
DEPENDENCY_PREFLIGHT_REMEDIATION_MATERIALIZATION,
));
}
if evidence
.iter()
.any(|item| item.status == DependencyPreflightStatus::Unknown)
{
return Some((
DEPENDENCY_PREFLIGHT_CODE_UNKNOWN,
DEPENDENCY_PREFLIGHT_REMEDIATION_UNKNOWN,
));
}
None
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub(super) struct DependencyPreflightCheck {
pub(super) root: String,
pub(super) manifest: String,
pub(super) required_path: String,
pub(super) required_kind: &'static str,
pub(super) is_primary: bool,
}
fn clean_relative_cargo_path(raw: &str) -> Option<PathBuf> {
let path = Path::new(raw);
if path.is_absolute() {
return None;
}
if path.components().any(|component| {
matches!(
component,
Component::ParentDir | Component::RootDir | Component::Prefix(_)
)
}) {
return None;
}
Some(path.to_path_buf())
}
fn table_path_value(table: &toml::Table, key: &str) -> Option<PathBuf> {
table
.get(key)
.and_then(toml::Value::as_str)
.and_then(clean_relative_cargo_path)
}
fn target_array_paths(table: &toml::Table, key: &str) -> Vec<PathBuf> {
table
.get(key)
.and_then(toml::Value::as_array)
.into_iter()
.flatten()
.filter_map(toml::Value::as_table)
.filter_map(|target| table_path_value(target, "path"))
.collect()
}
fn table_string_array(table: &toml::Table, key: &str) -> Vec<String> {
table
.get(key)
.and_then(toml::Value::as_array)
.into_iter()
.flatten()
.filter_map(toml::Value::as_str)
.map(str::to_string)
.collect()
}
fn package_auto_discovery_enabled(table: &toml::Table, key: &str) -> bool {
table
.get("package")
.and_then(toml::Value::as_table)
.and_then(|package| package.get(key))
.and_then(toml::Value::as_bool)
.unwrap_or(true)
}
fn insert_existing_relative_entrypoint(
paths: &mut std::collections::BTreeSet<PathBuf>,
package_root: &Path,
relative_path: impl Into<PathBuf>,
) {
let relative_path = relative_path.into();
if package_root.join(&relative_path).is_file() {
paths.insert(relative_path);
}
}
fn insert_auto_discovered_target_entrypoints(
paths: &mut std::collections::BTreeSet<PathBuf>,
package_root: &Path,
relative_dir: &str,
) {
let Ok(entries) = std::fs::read_dir(package_root.join(relative_dir)) else {
return;
};
for entry in entries.flatten() {
let Ok(file_type) = entry.file_type() else {
continue;
};
let file_name = entry.file_name();
if file_type.is_file() {
if entry.path().extension().is_some_and(|ext| ext == "rs") {
paths.insert(PathBuf::from(relative_dir).join(file_name));
}
} else if file_type.is_dir() {
insert_existing_relative_entrypoint(
paths,
package_root,
PathBuf::from(relative_dir).join(file_name).join("main.rs"),
);
}
}
}
pub(super) fn cargo_package_source_entrypoints(package_root: &Path) -> Vec<PathBuf> {
let manifest_path = package_root.join("Cargo.toml");
let Ok(contents) = std::fs::read_to_string(&manifest_path) else {
return Vec::new();
};
let Ok(table) = toml::from_str::<toml::Table>(&contents) else {
return Vec::new();
};
if !table.contains_key("package") {
return Vec::new();
}
let mut paths = std::collections::BTreeSet::<PathBuf>::new();
if let Some(lib_path) = table
.get("lib")
.and_then(toml::Value::as_table)
.and_then(|lib| table_path_value(lib, "path"))
{
paths.insert(lib_path);
} else if package_auto_discovery_enabled(&table, "autolib") || table.contains_key("lib") {
insert_existing_relative_entrypoint(&mut paths, package_root, "src/lib.rs");
}
if package_auto_discovery_enabled(&table, "autobins") {
insert_existing_relative_entrypoint(&mut paths, package_root, "src/main.rs");
insert_auto_discovered_target_entrypoints(&mut paths, package_root, "src/bin");
}
if package_auto_discovery_enabled(&table, "autoexamples") {
insert_auto_discovered_target_entrypoints(&mut paths, package_root, "examples");
}
if package_auto_discovery_enabled(&table, "autotests") {
insert_auto_discovered_target_entrypoints(&mut paths, package_root, "tests");
}
if package_auto_discovery_enabled(&table, "autobenches") {
insert_auto_discovered_target_entrypoints(&mut paths, package_root, "benches");
}
for key in ["bin", "example", "test", "bench"] {
paths.extend(target_array_paths(&table, key));
}
paths.into_iter().collect()
}
fn cargo_manifest_table(package_root: &Path) -> Option<toml::Table> {
let manifest_path = package_root.join("Cargo.toml");
let contents = std::fs::read_to_string(&manifest_path).ok()?;
toml::from_str::<toml::Table>(&contents).ok()
}
fn has_glob_meta(pattern: &str) -> bool {
pattern
.bytes()
.any(|byte| matches!(byte, b'*' | b'?' | b'['))
}
fn path_slash_string(path: &Path) -> Option<String> {
Some(
path.to_str()?
.trim_start_matches("./")
.replace(std::path::MAIN_SEPARATOR, "/"),
)
}
fn workspace_exclude_matches(relative_member: &Path, exclude_patterns: &[String]) -> bool {
let Some(relative_member) = path_slash_string(relative_member) else {
return false;
};
exclude_patterns.iter().any(|pattern| {
let Some(pattern_path) = clean_relative_cargo_path(pattern) else {
return false;
};
let Some(pattern_slash) = path_slash_string(&pattern_path) else {
return false;
};
if has_glob_meta(&pattern_slash) {
glob::Pattern::new(&pattern_slash)
.map(|compiled| compiled.matches(&relative_member))
.unwrap_or(false)
} else {
relative_member == pattern_slash
|| relative_member
.strip_prefix(&pattern_slash)
.is_some_and(|suffix| suffix.starts_with('/'))
}
})
}
fn workspace_member_manifest_paths(workspace_root: &Path) -> Vec<PathBuf> {
let Some(table) = cargo_manifest_table(workspace_root) else {
return Vec::new();
};
let Some(workspace) = table.get("workspace").and_then(toml::Value::as_table) else {
return Vec::new();
};
let members = table_string_array(workspace, "members");
let exclude_patterns = table_string_array(workspace, "exclude");
let mut manifests = std::collections::BTreeSet::<PathBuf>::new();
for member in members {
let Some(member_path) = clean_relative_cargo_path(&member) else {
continue;
};
if has_glob_meta(&member) {
let glob_path = workspace_root.join(&member_path).join("Cargo.toml");
let Some(glob_pattern) = glob_path.to_str() else {
continue;
};
let Ok(entries) = glob::glob(glob_pattern) else {
continue;
};
for entry in entries.flatten() {
if !entry.is_file() {
continue;
}
let Ok(relative_manifest) = entry.strip_prefix(workspace_root) else {
continue;
};
let Some(relative_member) = relative_manifest.parent() else {
continue;
};
if !workspace_exclude_matches(relative_member, &exclude_patterns) {
manifests.insert(relative_manifest.to_path_buf());
}
}
} else {
let relative_manifest = member_path.join("Cargo.toml");
if workspace_root.join(&relative_manifest).is_file()
&& !workspace_exclude_matches(&member_path, &exclude_patterns)
{
manifests.insert(relative_manifest);
}
}
}
manifests.into_iter().collect()
}
pub(super) fn cargo_workspace_member_source_entrypoints(workspace_root: &Path) -> Vec<PathBuf> {
let mut paths = std::collections::BTreeSet::<PathBuf>::new();
for manifest_path in workspace_member_manifest_paths(workspace_root) {
paths.insert(manifest_path.clone());
if let Some(member_root) = manifest_path.parent() {
paths.extend(
cargo_package_source_entrypoints(&workspace_root.join(member_root))
.into_iter()
.map(|entrypoint| member_root.join(entrypoint)),
);
}
}
paths.into_iter().collect()
}
pub(super) fn dependency_preflight_checks_for_entry(
entry: &SyncClosurePlanEntry,
) -> Vec<DependencyPreflightCheck> {
let remote_root = PathBuf::from(&entry.remote_root);
let manifest = remote_root.join("Cargo.toml").to_string_lossy().to_string();
let mut checks = Vec::new();
checks.push(DependencyPreflightCheck {
root: entry.remote_root.clone(),
manifest: manifest.clone(),
required_path: manifest.clone(),
required_kind: "manifest",
is_primary: entry.is_primary,
});
if entry.mode == SyncClosureMode::Full {
checks.extend(
cargo_package_source_entrypoints(&entry.local_root)
.into_iter()
.map(|relative_path| DependencyPreflightCheck {
root: entry.remote_root.clone(),
manifest: manifest.clone(),
required_path: remote_root
.join(relative_path)
.to_string_lossy()
.to_string(),
required_kind: "source_entrypoint",
is_primary: entry.is_primary,
}),
);
checks.extend(
cargo_workspace_member_source_entrypoints(&entry.local_root)
.into_iter()
.map(|relative_path| DependencyPreflightCheck {
root: entry.remote_root.clone(),
manifest: manifest.clone(),
required_path: remote_root
.join(relative_path)
.to_string_lossy()
.to_string(),
required_kind: "source_entrypoint",
is_primary: entry.is_primary,
}),
);
}
checks.sort_by(|left, right| {
(&left.required_path, left.required_kind).cmp(&(&right.required_path, right.required_kind))
});
checks.dedup_by(|left, right| {
left.required_path == right.required_path && left.required_kind == right.required_kind
});
checks
}
pub(super) fn synced_dependency_preflight_checks(
root_outcomes: &[(SyncClosurePlanEntry, SyncRootOutcome)],
) -> Vec<DependencyPreflightCheck> {
root_outcomes
.iter()
.filter(|(_, outcome)| matches!(outcome, SyncRootOutcome::Synced))
.flat_map(|(entry, _)| dependency_preflight_checks_for_entry(entry))
.collect()
}
pub(super) fn build_dependency_preflight_report(
worker: &WorkerConfig,
root_outcomes: &[(SyncClosurePlanEntry, SyncRootOutcome)],
present_paths: &std::collections::BTreeSet<String>,
missing_paths: &std::collections::BTreeSet<String>,
probe_failure: Option<&str>,
) -> DependencyPreflightReport {
let mut evidence = Vec::new();
let mut unknown_probe_samples = std::collections::BTreeSet::<(String, &'static str)>::new();
for (entry, outcome) in root_outcomes {
for check in dependency_preflight_checks_for_entry(entry) {
let (status, reason_code, detail) = match outcome {
SyncRootOutcome::Synced => {
if missing_paths.contains(&check.required_path) {
let detail = match check.required_kind {
"manifest" => "required Cargo.toml is missing on remote worker",
"source_entrypoint" => {
"required package source entrypoint is missing on remote worker"
}
_ => "required dependency path is missing on remote worker",
};
(
DependencyPreflightStatus::Missing,
DEPENDENCY_PREFLIGHT_CODE_MISSING,
format!("{detail}: {}", check.required_path),
)
} else if present_paths.contains(&check.required_path) {
(
DependencyPreflightStatus::Present,
DEPENDENCY_PREFLIGHT_CODE_PRESENT,
format!(
"required {} present after current sync: {}",
check.required_kind, check.required_path
),
)
} else {
let detail = probe_failure
.map(|failure| {
format!(
"dependency probe unavailable for {} under {}; sample required_path={}; additional unreported paths for this root/kind share this failure; {}",
check.required_kind,
check.root,
check.required_path,
failure
)
})
.unwrap_or_else(|| {
format!(
"probe output omitted status for synced required path: {}",
check.required_path
)
});
(
DependencyPreflightStatus::Unknown,
DEPENDENCY_PREFLIGHT_CODE_UNKNOWN,
detail,
)
}
}
SyncRootOutcome::Skipped { reason } => (
DependencyPreflightStatus::Stale,
DEPENDENCY_PREFLIGHT_CODE_STALE,
format!(
"dependency root skipped before verification for {}: {}",
check.required_path, reason
),
),
SyncRootOutcome::Failed { error } => (
DependencyPreflightStatus::Unknown,
DEPENDENCY_PREFLIGHT_CODE_UNKNOWN,
format!(
"dependency root sync failed before verification for {}: {}",
check.required_path, error
),
),
};
if probe_failure.is_some()
&& status == DependencyPreflightStatus::Unknown
&& matches!(outcome, SyncRootOutcome::Synced)
&& !unknown_probe_samples.insert((check.root.clone(), check.required_kind))
{
continue;
}
evidence.push(DependencyPreflightEvidence {
root: check.root,
manifest: check.manifest,
required_path: check.required_path,
required_kind: check.required_kind,
status,
reason_code,
detail,
is_primary: check.is_primary,
});
}
}
if evidence.is_empty() {
for (entry, outcome) in root_outcomes {
let manifest = PathBuf::from(&entry.remote_root)
.join("Cargo.toml")
.to_string_lossy()
.to_string();
let (status, reason_code, detail) = match outcome {
SyncRootOutcome::Synced => (
DependencyPreflightStatus::Unknown,
DEPENDENCY_PREFLIGHT_CODE_UNKNOWN,
"dependency preflight had no required paths to verify".to_string(),
),
SyncRootOutcome::Skipped { reason } => (
DependencyPreflightStatus::Stale,
DEPENDENCY_PREFLIGHT_CODE_STALE,
format!("dependency root skipped before verification: {}", reason),
),
SyncRootOutcome::Failed { error } => (
DependencyPreflightStatus::Unknown,
DEPENDENCY_PREFLIGHT_CODE_UNKNOWN,
format!("dependency root sync failed before verification: {}", error),
),
};
evidence.push(DependencyPreflightEvidence {
root: entry.remote_root.clone(),
manifest: manifest.clone(),
required_path: manifest,
required_kind: "manifest",
status,
reason_code,
detail,
is_primary: entry.is_primary,
});
}
}
let (verified, reason_code, remediation) = match dependency_preflight_failure_reason(&evidence)
{
Some((reason_code, remediation)) => (false, Some(reason_code), Some(remediation)),
None => (true, None, None),
};
DependencyPreflightReport {
schema_version: DEPENDENCY_PREFLIGHT_SCHEMA_VERSION,
worker: worker.id.as_str().to_string(),
verified,
reason_code,
remediation,
evidence,
}
}
pub(super) fn canonicalize_sync_root_for_plan(root: &Path, policy: &PathTopologyPolicy) -> PathBuf {
normalize_dependency_root_for_runtime(root, policy)
.or_else(|| std::fs::canonicalize(root).ok())
.unwrap_or_else(|| root.to_path_buf())
}
fn manifest_declares_workspace(manifest_path: &Path) -> bool {
let Ok(contents) = std::fs::read_to_string(manifest_path) else {
return false;
};
let Ok(table) = toml::from_str::<toml::Table>(&contents) else {
return false;
};
table.contains_key("workspace")
}
fn enclosing_workspace_root_for_sync_root(
root: &Path,
policy: &PathTopologyPolicy,
) -> Option<PathBuf> {
for candidate in root.ancestors() {
if !is_within_sync_topology(candidate, policy) {
break;
}
let manifest_path = candidate.join("Cargo.toml");
if manifest_path.is_file() && manifest_declares_workspace(&manifest_path) {
return (candidate != root).then(|| candidate.to_path_buf());
}
}
None
}
pub(super) fn workspace_metadata_sync_patterns() -> Vec<String> {
WORKSPACE_METADATA_SYNC_PATTERNS
.iter()
.map(|pattern| (*pattern).to_string())
.collect()
}
fn effective_sync_topology_roots(policy: &PathTopologyPolicy) -> Vec<PathBuf> {
let mut roots = Vec::new();
roots.push(policy.canonical_root().to_path_buf());
roots.push(policy.alias_root().to_path_buf());
if !policy.canonical_root().exists()
&& let Ok(alias_target) = std::fs::canonicalize(policy.alias_root())
{
roots.push(alias_target);
}
roots.sort();
roots.dedup();
roots
}
pub(super) fn is_within_sync_topology(path: &Path, policy: &PathTopologyPolicy) -> bool {
effective_sync_topology_roots(policy)
.into_iter()
.any(|root| path.starts_with(&root))
}
fn map_sync_root_to_remote_root(path: &Path, policy: &PathTopologyPolicy) -> String {
let remote_root = policy.canonical_root();
for root in effective_sync_topology_roots(policy) {
if let Ok(relative) = path.strip_prefix(&root) {
if relative.as_os_str().is_empty() {
return remote_root.to_string_lossy().to_string();
}
return remote_root.join(relative).to_string_lossy().to_string();
}
}
path.to_string_lossy().to_string()
}
pub(super) fn build_sync_closure_plan(
sync_roots: &[PathBuf],
normalized_project_root: &Path,
project_hash: &str,
topology_policy: &PathTopologyPolicy,
) -> Vec<SyncClosurePlanEntry> {
let mut ordered_entries = std::collections::BTreeSet::<(PathBuf, SyncClosureMode)>::new();
let primary_root = canonicalize_sync_root_for_plan(normalized_project_root, topology_policy);
for root in sync_roots {
let canonicalized = canonicalize_sync_root_for_plan(root, topology_policy);
if !is_within_sync_topology(&canonicalized, topology_policy) {
warn!(
"Dependency root {} (canonicalized: {}) is outside allowed topology ({} / {}); skipping from sync closure",
root.display(),
canonicalized.display(),
topology_policy.canonical_root().display(),
topology_policy.alias_root().display(),
);
continue;
}
if canonicalized != primary_root && canonicalized.starts_with(&primary_root) {
continue;
}
ordered_entries.insert((canonicalized.clone(), SyncClosureMode::Full));
if let Some(workspace_root) =
enclosing_workspace_root_for_sync_root(&canonicalized, topology_policy)
{
ordered_entries.insert((workspace_root, SyncClosureMode::WorkspaceMetadata));
}
}
ordered_entries.insert((primary_root.clone(), SyncClosureMode::Full));
let full_roots: Vec<PathBuf> = ordered_entries
.iter()
.filter(|(_, mode)| *mode == SyncClosureMode::Full)
.map(|(root, _)| root.clone())
.collect();
for root in full_roots {
ordered_entries.remove(&(root, SyncClosureMode::WorkspaceMetadata));
}
ordered_entries
.into_iter()
.map(|(root, mode)| {
let is_primary = root == primary_root && mode == SyncClosureMode::Full;
let root_hash = if is_primary {
project_hash.to_string()
} else {
compute_project_hash_with_dependency_roots_and_policy(&root, &[], topology_policy)
};
SyncClosurePlanEntry {
remote_root: map_sync_root_to_remote_root(&root, topology_policy),
project_id: project_id_from_path(&root),
root_hash,
is_primary,
mode,
local_root: root,
}
})
.collect()
}
pub(super) fn build_owned_sync_closure_layout(
plan: &[SyncClosurePlanEntry],
topology_policy: &PathTopologyPolicy,
remote_base: &str,
legacy_pool_name: &str,
) -> Result<OwnedSyncClosureLayout, OwnedSyncClosureLayoutError> {
use OwnedSyncClosureLayoutError as Error;
let base = rch_common::types::validate_remote_base(remote_base).map_err(Error::InvalidBase)?;
if base.chars().any(char::is_control)
|| base.contains('\\')
|| base.split('/').any(|part| part == "." || part == "..")
{
return Err(Error::InvalidBase("invalid path component".into()));
}
let base: PathBuf = Path::new(&base).components().collect();
let base = base
.to_str()
.ok_or_else(|| Error::InvalidBase("path is not UTF-8".into()))?;
let (pool_prefix, _) = legacy_pool_name
.rsplit_once("-pool-")
.filter(|(prefix, key)| {
prefix.starts_with(".rch-target-")
&& prefix.len() > ".rch-target-".len()
&& prefix[".rch-target-".len()..]
.bytes()
.all(|byte| byte.is_ascii_alphanumeric() || b"-_".contains(&byte))
&& key.len() == 32
&& key.bytes().all(|byte| byte.is_ascii_hexdigit())
})
.ok_or(Error::InvalidPoolName)?;
if plan.iter().filter(|entry| entry.is_primary).count() != 1
|| plan
.iter()
.any(|entry| entry.is_primary && entry.mode != SyncClosureMode::Full)
{
return Err(Error::InvalidPlan(
"exactly one full primary root is required",
));
}
let mut topology_root = None;
let mut entries = Vec::with_capacity(plan.len());
for entry in plan {
let normalized = normalize_project_path_with_policy(&entry.local_root, topology_policy)
.map_err(|error| Error::InvalidRoot {
root: entry.local_root.clone(),
reason: error.to_string(),
})?;
if topology_root
.as_ref()
.is_some_and(|root| root != normalized.canonical_root())
{
return Err(Error::InvalidPlan(
"canonical topology changed during planning",
));
}
topology_root = Some(normalized.canonical_root().to_path_buf());
let relative = normalized
.canonical_path()
.strip_prefix(normalized.canonical_root())
.map_err(|_| Error::InvalidPlan("root escaped its normalized topology"))?;
let literal = relative.to_str().ok_or_else(|| Error::InvalidRoot {
root: entry.local_root.clone(),
reason: "path is not UTF-8".into(),
})?;
if !normalized.canonical_path().is_dir()
|| literal.chars().any(char::is_control)
|| literal.contains('\\')
|| relative
.components()
.any(|part| !matches!(part, std::path::Component::Normal(_)))
{
return Err(Error::InvalidRoot {
root: entry.local_root.clone(),
reason: "root must be a directory with literal path components".into(),
});
}
let mut entry = entry.clone();
entry.local_root = normalized.canonical_path().to_path_buf();
entries.push((relative.to_path_buf(), entry));
}
entries.sort_by(|(left, _), (right, _)| left.cmp(right));
for (index, (ancestor, entry)) in entries.iter().enumerate() {
for (descendant, other) in &entries[index + 1..] {
let Ok(relative) = descendant.strip_prefix(ancestor) else {
continue;
};
let metadata_overlap = relative.components().next().is_some_and(|part| {
matches!(
part.as_os_str().to_str(),
Some(
".cargo"
| "Cargo.toml"
| "Cargo.lock"
| "rust-toolchain"
| "rust-toolchain.toml"
)
)
});
if relative.as_os_str().is_empty()
|| entry.mode == SyncClosureMode::Full
|| metadata_overlap
{
return Err(Error::OverlappingRoots {
first: entry.local_root.clone(),
second: other.local_root.clone(),
});
}
}
}
fn field(hasher: &mut blake3::Hasher, value: &[u8]) {
hasher.update(&(value.len() as u64).to_le_bytes());
hasher.update(value);
}
let mut hasher = blake3::Hasher::new();
hasher.update(b"rch-source-content-pair-layout-v1\0");
field(&mut hasher, base.as_bytes());
field(&mut hasher, legacy_pool_name.as_bytes());
for (relative, entry) in &entries {
field(
&mut hasher,
relative.to_str().expect("validated UTF-8 path").as_bytes(),
);
field(
&mut hasher,
match entry.mode {
SyncClosureMode::Full => b"full",
SyncClosureMode::WorkspaceMetadata => b"workspace_metadata",
},
);
field(&mut hasher, &[u8::from(entry.is_primary)]);
}
let key = hasher.finalize().to_hex()[..32].to_string();
let source_root = format!("{base}/source-content-pair-{key}");
let projected_root = Path::new(&source_root).join("projects");
let plan = entries
.into_iter()
.map(|(relative, mut entry)| {
let destination = if relative.as_os_str().is_empty() {
projected_root.clone()
} else {
projected_root.join(relative)
};
entry.remote_root = destination.to_string_lossy().into_owned();
entry
})
.collect();
Ok(OwnedSyncClosureLayout {
source_root,
pooled_target_name: format!("{pool_prefix}-pool-{key}"),
plan,
})
}
pub(super) fn build_sync_closure_manifest(
plan: &[SyncClosurePlanEntry],
normalized_project_root: &Path,
) -> SyncClosureManifest {
let generated_at_unix_ms = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|duration| duration.as_millis() as i64)
.unwrap_or_default();
let entries = plan
.iter()
.enumerate()
.map(|(idx, entry)| SyncClosureManifestEntry {
order: idx + 1,
local_root: entry.local_root.to_string_lossy().to_string(),
remote_root: entry.remote_root.clone(),
project_id: entry.project_id.clone(),
root_hash: entry.root_hash.clone(),
is_primary: entry.is_primary,
mode: entry.mode,
})
.collect();
SyncClosureManifest {
schema_version: "rch.sync_closure_manifest.v2",
generated_at_unix_ms,
project_root: normalized_project_root.to_string_lossy().to_string(),
entries,
}
}
pub(super) async fn verify_remote_dependency_manifests(
worker: &WorkerConfig,
root_outcomes: &[(SyncClosurePlanEntry, SyncRootOutcome)],
reporter: &HookReporter,
source_identity: Option<&str>,
) -> anyhow::Result<()> {
if should_skip_remote_preflight(worker) {
reporter.verbose("[RCH] remote dependency preflight skipped in mock mode");
return Ok(());
}
if root_outcomes.is_empty() {
return Ok(());
}
let synced_checks = synced_dependency_preflight_checks(root_outcomes);
let mut present_paths = std::collections::BTreeSet::new();
let mut missing_paths = std::collections::BTreeSet::new();
let mut probe_failure: Option<String> = None;
if let Some(verify_cmd) = build_remote_dependency_preflight_command(&synced_checks) {
let remote_command = source_identity
.map(|identity| super::ssh::wrap_remote_source_activity("sh -s", identity))
.transpose()?
.unwrap_or_else(|| "sh -s".to_owned());
match run_offload_ssh_command_with_stdin(
worker,
&remote_command,
verify_cmd.as_bytes(),
dependency_preflight_timeout(synced_checks.len()),
)
.await
{
Ok(output) => {
let stdout = String::from_utf8_lossy(&output.stdout).trim().to_string();
let stderr = String::from_utf8_lossy(&output.stderr).trim().to_string();
let (present, missing) = parse_dependency_preflight_probe_output(&stdout);
let expected_missing_exit = output.status.code() == Some(43) && !missing.is_empty();
if output.status.success() || expected_missing_exit {
present_paths.extend(present);
missing_paths.extend(missing);
} else {
probe_failure = Some(format!(
"probe exited with status {:?}; stdout='{}'; stderr='{}'",
output.status.code(),
stdout,
stderr
));
}
}
Err(err) => {
probe_failure = Some(err.to_string());
}
}
}
let report = build_dependency_preflight_report(
worker,
root_outcomes,
&present_paths,
&missing_paths,
probe_failure.as_deref(),
);
let report_json = serde_json::to_string(&report).unwrap_or_else(|err| {
format!(
"{{\"schema_version\":\"{}\",\"verified\":false,\"reason_code\":\"{}\",\"serialization_error\":\"{}\"}}",
DEPENDENCY_PREFLIGHT_SCHEMA_VERSION, DEPENDENCY_PREFLIGHT_CODE_UNKNOWN, err
)
});
reporter.verbose(&format!(
"[RCH] dependency preflight report: {}",
report_json
));
if report.verified {
reporter.verbose(&format!(
"[RCH] remote dependency preflight verified {} roots on {}",
report.evidence.len(),
worker.id
));
return Ok(());
}
let failure = DependencyPreflightFailure::from_report(report);
let evidence_summary = failure.evidence_summary();
warn!(
"Remote dependency preflight blocked remote execution on {} [{}] remediation='{}' evidence='{}'",
worker.id, failure.reason_code, failure.remediation, evidence_summary
);
reporter.verbose(&format!(
"[RCH] dependency preflight remediation [{}]: {}",
failure.reason_code, failure.remediation
));
Err(failure.into())
}
pub(super) fn build_remote_dependency_preflight_command(
checks: &[DependencyPreflightCheck],
) -> Option<String> {
if checks.is_empty() {
return None;
}
let commands = checks
.iter()
.map(|check| {
let escaped = shell_escape::escape(check.required_path.clone().into());
escaped.to_string()
})
.collect::<Vec<_>>()
.join(" ");
Some(format!(
"missing=0; for required in {commands}; do if [ -f \"$required\" ]; then printf 'RCH_DEP_PRESENT:%s\\n' \"$required\"; else printf 'RCH_DEP_MISSING:%s\\n' \"$required\"; missing=1; fi; done; if [ \"$missing\" -ne 0 ]; then exit 43; fi; echo RCH_REMOTE_DEPENDENCIES_OK"
))
}
pub(super) fn dependency_preflight_timeout(check_count: usize) -> Duration {
let windows = check_count
.div_ceil(DEPENDENCY_PREFLIGHT_PATHS_PER_TIMEOUT_WINDOW)
.max(1);
Duration::from_secs(20).saturating_mul(u32::try_from(windows).unwrap_or(u32::MAX))
}
#[cfg(all(test, unix))]
mod owned_layout_tests {
use super::*;
use rch_common::cargo_path_deps::{
SelectedCargoEntry, validate_selected_cargo_path, validate_selected_cargo_tree_at,
};
use rch_common::test_guard;
use std::collections::BTreeMap;
use std::fs;
const POOL: &str = ".rch-target-worker-pool-0123456789abcdef0123456789abcdef";
fn fixture() -> (
tempfile::TempDir,
PathTopologyPolicy,
Vec<SyncClosurePlanEntry>,
) {
let temp = tempfile::tempdir().unwrap();
let root = temp.path().join("projects");
fs::create_dir_all(root.join("app/src")).unwrap();
fs::create_dir_all(root.join("foreign/crates/member/src")).unwrap();
let root = fs::canonicalize(root).unwrap();
let policy = PathTopologyPolicy::new(root.clone(), root.clone());
fs::write(
root.join("foreign/Cargo.toml"),
"[workspace]\nmembers = [\"crates/member\"]\n[workspace.package]\nedition = \"2021\"\n",
)
.unwrap();
let plan = build_sync_closure_plan(
&[root.join("app"), root.join("foreign/crates/member")],
&root.join("app"),
"source-stamp-a",
&policy,
);
(temp, policy, plan)
}
#[test]
fn owned_layout_preserves_siblings_workspace_metadata_and_manifest_provenance() {
let _guard = test_guard!();
let (temp, policy, plan) = fixture();
let root = policy.canonical_root();
let files = [
(
"app/Cargo.toml",
"[package]\nname = \"app\"\nversion = \"0.1.0\"\nedition = \"2021\"\n[dependencies]\nmember = { path = \"../foreign/crates/member\" }\n",
),
(
"app/src/lib.rs",
"pub fn value() -> u32 { member::value() }\n",
),
(
"foreign/Cargo.toml",
"[workspace]\nmembers = [\"crates/member\"]\n[workspace.package]\nedition = \"2021\"\n",
),
(
"foreign/crates/member/Cargo.toml",
"[package]\nname = \"member\"\nversion = \"0.1.0\"\nedition.workspace = true\n",
),
(
"foreign/crates/member/src/lib.rs",
"pub fn value() -> u32 { 7 }\n",
),
];
for (relative, contents) in files {
fs::write(root.join(relative), contents).unwrap();
}
fs::create_dir_all(root.join("app/.git")).unwrap();
fs::write(root.join("app/.git/HEAD"), "canonical-checkout-sentinel\n").unwrap();
let base = temp.path().join("worker/staging");
let layout =
build_owned_sync_closure_layout(&plan, &policy, base.to_str().unwrap(), POOL).unwrap();
assert_eq!(layout.plan.len(), 3);
assert!(layout.plan.iter().any(|entry| {
entry.mode == SyncClosureMode::WorkspaceMetadata
&& entry.remote_root == format!("{}/projects/foreign", layout.source_root)
}));
let mut selected = BTreeMap::new();
for (relative, contents) in files {
let local = root.join(relative);
let entry = layout
.plan
.iter()
.filter(|entry| local.starts_with(&entry.local_root))
.max_by_key(|entry| entry.local_root.components().count())
.unwrap();
let destination =
Path::new(&entry.remote_root).join(local.strip_prefix(&entry.local_root).unwrap());
fs::create_dir_all(destination.parent().unwrap()).unwrap();
fs::copy(&local, &destination).unwrap();
assert_eq!(fs::read_to_string(&destination).unwrap(), contents);
selected.insert(
destination
.strip_prefix(&layout.source_root)
.unwrap()
.to_path_buf(),
SelectedCargoEntry::File(
relative
.ends_with("Cargo.toml")
.then(|| contents.to_owned()),
),
);
}
let selected_manifest = Path::new("projects/app/Cargo.toml");
validate_selected_cargo_tree_at(&selected, selected_manifest).unwrap();
assert_eq!(
validate_selected_cargo_path(
&selected,
Path::new("projects/app/../foreign/crates/member/Cargo.toml"),
)
.unwrap(),
Path::new("projects/foreign/crates/member/Cargo.toml")
);
let manifest = build_sync_closure_manifest(&layout.plan, &root.join("app"));
for entry in &manifest.entries {
let original = plan
.iter()
.find(|original| original.local_root == Path::new(&entry.local_root))
.unwrap();
assert_eq!(entry.root_hash, original.root_hash);
assert_eq!(entry.project_id, original.project_id);
assert_eq!(entry.mode, original.mode);
assert_eq!(entry.is_primary, original.is_primary);
assert!(entry.remote_root.starts_with(&layout.source_root));
}
assert_eq!(
fs::read_to_string(root.join("app/.git/HEAD")).unwrap(),
"canonical-checkout-sentinel\n"
);
assert!(
!Path::new(&layout.source_root)
.join("projects/app/.git")
.exists()
);
selected.insert(
selected_manifest.to_path_buf(),
SelectedCargoEntry::File(Some(format!(
"[package]\nname = \"app\"\nversion = \"0.1.0\"\n[dependencies]\nmember = {{ path = {:?} }}\n",
root.join("foreign/crates/member").to_str().unwrap()
))),
);
assert!(validate_selected_cargo_tree_at(&selected, selected_manifest).is_err());
}
#[test]
fn owned_layout_reuses_paths_across_source_edits_and_plan_order() {
let _guard = test_guard!();
let (_temp, policy, plan) = fixture();
let first = build_owned_sync_closure_layout(&plan, &policy, "/tmp/rch", POOL).unwrap();
let mut changed = plan.clone();
changed.reverse();
for entry in &mut changed {
entry.root_hash = "source-stamp-b".into();
}
fs::write(
policy.canonical_root().join("app/src/lib.rs"),
"pub fn changed() {}\n",
)
.unwrap();
let next = build_owned_sync_closure_layout(&changed, &policy, "/tmp//rch/", POOL).unwrap();
assert_eq!(first.source_root, next.source_root);
assert_eq!(first.pooled_target_name, next.pooled_target_name);
assert_ne!(first.pooled_target_name, POOL);
for (before, after) in first.plan.iter().zip(&next.plan) {
assert_eq!(before.local_root, after.local_root);
assert_eq!(before.remote_root, after.remote_root);
assert_eq!(after.root_hash, "source-stamp-b");
}
}
#[test]
fn owned_layout_key_binds_target_dimensions_worker_base_and_source_projection() {
let _guard = test_guard!();
let (_temp, policy, plan) = fixture();
let first = build_owned_sync_closure_layout(&plan, &policy, "/tmp/rch", POOL).unwrap();
let other_pool = ".rch-target-worker-pool-fedcba9876543210fedcba9876543210";
let changed_pool =
build_owned_sync_closure_layout(&plan, &policy, "/tmp/rch", other_pool).unwrap();
let changed_base =
build_owned_sync_closure_layout(&plan, &policy, "/data/tmp/rch", POOL).unwrap();
let mut selected = plan.clone();
selected.retain(|entry| entry.is_primary);
let changed_projection =
build_owned_sync_closure_layout(&selected, &policy, "/tmp/rch", POOL).unwrap();
for changed in [changed_pool, changed_base, changed_projection] {
assert_ne!(first.source_root, changed.source_root);
assert_ne!(first.pooled_target_name, changed.pooled_target_name);
}
}
#[test]
fn owned_layout_normalizes_aliases_and_a_symlinked_canonical_root() {
let _guard = test_guard!();
let (temp, policy, plan) = fixture();
let configured = temp.path().join("configured-projects");
let alias = temp.path().join("alias-projects");
std::os::unix::fs::symlink(policy.canonical_root(), &configured).unwrap();
std::os::unix::fs::symlink(policy.canonical_root(), &alias).unwrap();
let aliased_policy = PathTopologyPolicy::new(configured, alias.clone());
let canonical =
build_owned_sync_closure_layout(&plan, &aliased_policy, "/tmp/rch", POOL).unwrap();
let mut aliased = plan;
for entry in &mut aliased {
entry.local_root = alias.join(
entry
.local_root
.strip_prefix(policy.canonical_root())
.unwrap(),
);
}
let aliases =
build_owned_sync_closure_layout(&aliased, &aliased_policy, "/tmp/rch", POOL).unwrap();
assert_eq!(canonical, aliases);
}
#[test]
fn owned_layout_rejects_ambiguous_authority_and_invalid_roots() {
let _guard = test_guard!();
let (temp, policy, plan) = fixture();
for base in [
"relative",
"/tmp",
"/tmp/./rch",
"/tmp/rch/../other",
"/tmp/rch\nother",
"/tmp/rch\\other",
] {
assert!(
matches!(
build_owned_sync_closure_layout(&plan, &policy, base, POOL),
Err(OwnedSyncClosureLayoutError::InvalidBase(_))
),
"{base:?}"
);
}
for pool in [
".rch-target-worker-job-1",
"../pool",
".rch-target-worker-pool-short",
".rch-target-unsafe/name-pool-0123456789abcdef0123456789abcdef",
] {
assert!(
matches!(
build_owned_sync_closure_layout(&plan, &policy, "/tmp/rch", pool),
Err(OwnedSyncClosureLayoutError::InvalidPoolName)
),
"{pool:?}"
);
}
assert!(matches!(
build_owned_sync_closure_layout(&[], &policy, "/tmp/rch", POOL),
Err(OwnedSyncClosureLayoutError::InvalidPlan(_))
));
let mut invalid = plan.clone();
invalid
.iter_mut()
.find(|entry| entry.is_primary)
.unwrap()
.local_root = temp.path().to_path_buf();
assert!(matches!(
build_owned_sync_closure_layout(&invalid, &policy, "/tmp/rch", POOL),
Err(OwnedSyncClosureLayoutError::InvalidRoot { .. })
));
let mut duplicate = plan.clone();
let mut extra = duplicate[0].clone();
extra.is_primary = false;
duplicate.push(extra);
assert!(matches!(
build_owned_sync_closure_layout(&duplicate, &policy, "/tmp/rch", POOL),
Err(OwnedSyncClosureLayoutError::OverlappingRoots { .. })
));
let mut overlapping = plan.clone();
overlapping
.iter_mut()
.find(|entry| entry.mode == SyncClosureMode::WorkspaceMetadata)
.unwrap()
.mode = SyncClosureMode::Full;
assert!(matches!(
build_owned_sync_closure_layout(&overlapping, &policy, "/tmp/rch", POOL),
Err(OwnedSyncClosureLayoutError::OverlappingRoots { .. })
));
}
#[test]
fn owned_layout_handles_a_primary_at_the_topology_root() {
let _guard = test_guard!();
let (_temp, policy, _) = fixture();
let plan = build_sync_closure_plan(&[], policy.canonical_root(), "topology-root", &policy);
let layout = build_owned_sync_closure_layout(&plan, &policy, "/tmp/rch", POOL).unwrap();
assert_eq!(layout.plan.len(), 1);
assert_eq!(
layout.plan[0].remote_root,
format!("{}/projects", layout.source_root)
);
}
}
#[cfg(all(test, unix))]
mod stdin_tests {
use super::*;
use rch_common::test_guard;
#[tokio::test]
async fn dependency_preflight_streams_large_batches() {
use std::os::unix::fs::PermissionsExt as _;
const DIR_ENV: &str = "RCH_DEPENDENCY_STDIN_TEST_DIR";
const ROOT_COUNT: usize = 128;
let _guard = test_guard!();
if let Some(dir) = std::env::var_os(DIR_ENV) {
let dir = PathBuf::from(dir);
let project = dir.join("project with ' quote");
let worker = WorkerConfig {
id: WorkerId::new("dependency-stdin-test"),
host: "dependency-stdin.invalid".into(),
user: "ubuntu".into(),
..WorkerConfig::default()
};
assert!(!should_skip_remote_preflight(&worker));
let outcomes: Vec<_> = (0..ROOT_COUNT)
.map(|index| {
let remote_root = if index == ROOT_COUNT - 1 {
project.clone()
} else {
let mut root = dir.join(format!("absent-{index}"));
for _ in 0..20 {
root.push("r".repeat(96));
}
root
};
(
SyncClosurePlanEntry {
local_root: project.clone(),
remote_root: remote_root.to_string_lossy().into_owned(),
project_id: format!("fixture-{index}"),
root_hash: "fixture".into(),
is_primary: index == ROOT_COUNT - 1,
mode: SyncClosureMode::Full,
},
SyncRootOutcome::Synced,
)
})
.collect();
let checks = synced_dependency_preflight_checks(&outcomes);
assert_eq!(checks.len(), ROOT_COUNT * 2);
let command = build_remote_dependency_preflight_command(&checks).unwrap();
assert!(command.len() > 256 * 1024);
let reporter = HookReporter::new(OutputVisibility::Summary);
let error = verify_remote_dependency_manifests(&worker, &outcomes, &reporter, None)
.await
.expect_err("absent fixture paths must remain missing, not pass preflight");
let report = error
.downcast::<DependencyPreflightFailure>()
.expect("typed dependency failure")
.report;
assert!(!report.verified);
assert_eq!(report.evidence.len(), ROOT_COUNT * 2);
assert_eq!(
report
.evidence
.iter()
.filter(|item| item.status == DependencyPreflightStatus::Missing)
.count(),
(ROOT_COUNT - 1) * 2,
"E2BIG must not turn missing files into unknown transport failures"
);
assert_eq!(
report
.evidence
.iter()
.filter(|item| item.status == DependencyPreflightStatus::Present)
.count(),
2,
"the final root proves the entire stdin script was processed"
);
assert_eq!(
std::fs::read_to_string(dir.join("ssh-calls")).unwrap(),
"probe\n",
"oversized multi-root checks must use exactly one SSH connection"
);
verify_remote_dependency_manifests(
&worker,
&outcomes[ROOT_COUNT - 1..],
&reporter,
None,
)
.await
.expect("healthy paths must still pass preflight");
for code in [1, 42, 43, 255] {
std::fs::write(dir.join("exit-after-output"), code.to_string()).unwrap();
let failure = verify_remote_dependency_manifests(
&worker,
&outcomes[ROOT_COUNT - 1..],
&reporter,
None,
)
.await
.expect_err("failed transport must not verify complete positive output")
.downcast::<DependencyPreflightFailure>()
.expect("typed transport verification failure");
assert_eq!(failure.reason_code, DEPENDENCY_PREFLIGHT_CODE_UNKNOWN);
assert!(!failure.report.verified);
assert_eq!(failure.report.evidence.len(), 2);
assert!(failure.report.evidence.iter().all(|item| {
item.status == DependencyPreflightStatus::Unknown
&& item.detail.contains(&format!("status Some({code})"))
}));
}
std::fs::write(dir.join("finished"), "ok").unwrap();
return;
}
let dir = tempfile::tempdir().unwrap();
let project = dir.path().join("project with ' quote");
std::fs::create_dir_all(project.join("src")).unwrap();
std::fs::write(
project.join("Cargo.toml"),
"[package]\nname = \"preflight-fixture\"\nversion = \"0.1.0\"\n",
)
.unwrap();
std::fs::write(project.join("src/lib.rs"), "pub fn fixture() {}\n").unwrap();
let bin = dir.path().join("bin");
std::fs::create_dir(&bin).unwrap();
let ssh = bin.join("ssh");
std::fs::write(
&ssh,
r#"#!/bin/sh
set -eu
LC_ALL=C
export LC_ALL
printf 'probe\n' >> "$RCH_DEPENDENCY_STDIN_TEST_DIR/ssh-calls"
remote=''
for arg; do
[ "${#arg}" -lt 131072 ] || exit 90
remote=$arg
done
[ "$remote" = "sh -lc 'sh -s'" ] || exit 91
if [ -f "$RCH_DEPENDENCY_STDIN_TEST_DIR/exit-after-output" ]; then
/bin/sh -c "$remote"
exit "$(cat "$RCH_DEPENDENCY_STDIN_TEST_DIR/exit-after-output")"
fi
exec /bin/sh -c "$remote"
"#,
)
.unwrap();
std::fs::set_permissions(&ssh, std::fs::Permissions::from_mode(0o755)).unwrap();
let old_path = std::env::var_os("PATH").unwrap_or_default();
let path =
std::env::join_paths(std::iter::once(bin).chain(std::env::split_paths(&old_path)))
.unwrap();
let name = concat!(
module_path!(),
"::dependency_preflight_streams_large_batches"
);
let name = name.split_once("::").unwrap().1;
let output = timeout(
Duration::from_secs(60),
Command::new(std::env::current_exe().unwrap())
.args(["--exact", name, "--nocapture"])
.env(DIR_ENV, dir.path())
.env("PATH", path)
.env("RCH_MOCK_SSH", "0")
.kill_on_drop(true)
.output(),
)
.await
.expect("dependency stdin regression timed out")
.expect("start isolated dependency stdin regression");
assert!(
output.status.success(),
"stdout={} stderr={}",
String::from_utf8_lossy(&output.stdout),
String::from_utf8_lossy(&output.stderr)
);
assert!(
dir.path().join("finished").is_file(),
"child test did not run"
);
}
}