const SOURCE_CLAIM_REGISTRY: &str = "/tmp/rch-source-authority-locks/claims-v1";
pub const REAP_GLOBS: &[&str] = &[
".rch-target-*-job-*",
".rch-target-*-pid-*",
".rch-target-*-pool-*",
];
pub const CARGO_CACHE_GLOB: &str = "rch-cargo-cache-*";
pub const MIN_CARGO_CACHE_IDLE_MINUTES: u64 = 24 * 60;
#[must_use]
pub fn pooled_idle_minutes_from_hours(idle_hours: u32) -> Option<u64> {
(idle_hours != 0).then(|| (u64::from(idle_hours) * 60).max(MIN_POOLED_IDLE_MINUTES))
}
#[must_use]
pub fn cargo_cache_idle_minutes_from_days(idle_days: u32) -> Option<u64> {
(idle_days != 0).then(|| (u64::from(idle_days) * 24 * 60).max(MIN_CARGO_CACHE_IDLE_MINUTES))
}
pub fn is_safe_reap_path(s: &str) -> bool {
s.starts_with('/')
&& s.matches('/').count() >= 2
&& !s.contains("..")
&& s.len() <= 4096
&& s.chars().all(is_safe_reap_char)
}
pub fn is_safe_reap_base(s: &str) -> bool {
s.starts_with('/')
&& s.matches('/').count() >= 1
&& s.trim_end_matches('/').len() > 1
&& !s.contains("..")
&& s.len() <= 4096
&& s.chars().all(is_safe_reap_char)
}
pub fn is_safe_reap_token(s: &str) -> bool {
!s.is_empty()
&& s != "."
&& s != ".."
&& !s.contains('/')
&& s.len() <= 255
&& s.chars().all(is_safe_reap_char)
}
pub fn is_safe_reap_char(c: char) -> bool {
c.is_ascii_alphanumeric() || matches!(c, '/' | '-' | '_' | '.')
}
pub const MIN_IDLE_HOURS: u32 = 1;
pub fn idle_minutes_from_hours(idle_hours: u32) -> u64 {
u64::from(idle_hours.max(MIN_IDLE_HOURS)) * 60
}
pub fn reap_loop_body(
idle_minutes: u64,
exclude_token: Option<&str>,
removed_counter: &str,
freed_kb: &str,
) -> String {
reap_loop_body_with_event(idle_minutes, exclude_token, removed_counter, freed_kb, None)
}
pub fn reap_loop_body_with_event(
idle_minutes: u64,
exclude_token: Option<&str>,
removed_counter: &str,
freed_kb: &str,
trigger: Option<&str>,
) -> String {
reap_loop_body_with_source_gate(
idle_minutes,
exclude_token,
removed_counter,
freed_kb,
trigger,
false,
)
}
fn reap_loop_body_with_source_gate(
idle_minutes: u64,
exclude_token: Option<&str>,
removed_counter: &str,
freed_kb: &str,
trigger: Option<&str>,
source_gate: bool,
) -> String {
let exclude = match exclude_token {
Some(tok) => format!("[ \"$d\" = \"{tok}\" ] && continue; "),
None => String::new(),
};
let (claim, unlock) = if source_gate {
(
format!(
"if ! __gc_source_begin \"$d\"; then printf 'RCH_GC_SKIP {} source-ownership %s\\n' \"$d\"; continue; fi; ",
trigger.unwrap_or("ttl")
),
"__gc_source_end; ",
)
} else {
(String::new(), "")
};
let (size_capture, removal) = if removed_counter.is_empty() || freed_kb.is_empty() {
(String::new(), "rm -rf -- \"$d\" 2>/dev/null;".to_string())
} else {
let removal_body = match trigger {
Some(tag) => format!(
"if __rmerr=$(rm -rf -- \"$d\" 2>&1); then \
{removed_counter}=$(({removed_counter} + 1)); {freed_kb}=$(({freed_kb} + sz)); \
printf 'RCH_REAP_RM %s {tag} %s\\n' \"$sz\" \"$d\"; \
else \
printf 'RCH_REAP_ERR {tag} %s :: %s\\n' \"$d\" \"$(printf '%s' \"$__rmerr\" | head -1)\"; \
fi;"
),
None => format!(
"if rm -rf -- \"$d\" 2>/dev/null; then {removed_counter}=$(({removed_counter} + 1)); {freed_kb}=$(({freed_kb} + sz)); fi;"
),
};
(
"sz=$(du -sk \"$d\" 2>/dev/null | awk '{print $1}'); [ -z \"$sz\" ] && sz=0; "
.to_string(),
removal_body,
)
};
format!(
"[ -d \"$d\" ] || continue; \
{exclude}\
{claim}\
if find \"$d\" -mmin -{idle_minutes} -print -quit 2>/dev/null | grep -q .; then {unlock}continue; fi; \
{size_capture}{removal}{unlock}"
)
}
fn source_claim_gate_fragment(registry: &str) -> String {
const GATE: &str = r#"
trap '' HUP;
__gc_source_realpath() {
__gc_resolved=$(realpath -m -- "$1" && printf '.') || return 1;
__gc_resolved=${__gc_resolved%.}; __gc_resolved=${__gc_resolved%?};
__gc_lines=$(printf '%s\n' "$__gc_resolved" | wc -l);
__gc_cr=$(printf '\r');
[ "$__gc_lines" -eq 1 ] || return 1;
case "$__gc_resolved" in /*) ;; *) return 1;; esac;
case "$__gc_resolved" in *"$__gc_cr"*) return 1;; esac;
};
__gc_source_records_free() {
__gc_source_realpath "$1" || return 1; __gc_candidate=$__gc_resolved;
for __gc_record in "$__gc_source_registry"/*.claim "$__gc_source_registry"/*.pending; do
[ -e "$__gc_record" ] || [ -L "$__gc_record" ] || continue;
[ ! -L "$__gc_record" ] && [ -f "$__gc_record" ] || return 1;
__gc_name=${__gc_record##*/}; __gc_token=${__gc_name%%.*};
__gc_digest=${__gc_name#*.}; __gc_digest=${__gc_digest%%.*};
case "$__gc_token" in ''|*[!a-f0-9-]*) return 1;; esac;
case "$__gc_digest" in *[!a-f0-9]*) return 1;; esac;
[ "${#__gc_digest}" -eq 64 ] || return 1;
__gc_size=$(wc -c < "$__gc_record") || return 1;
[ "$__gc_size" -gt 0 ] && [ "$__gc_size" -le 33554432 ] || return 1;
__gc_actual=$(sha256sum -- "$__gc_record") || return 1;
[ "${__gc_actual%% *}" = "$__gc_digest" ] || return 1;
while IFS= read -r __gc_root || [ -n "$__gc_root" ]; do
case "$__gc_root" in /*) ;; *) return 1;; esac;
case "$__gc_root" in *//*|*/./*|*/../*|*/.|*/..) return 1;; esac;
[ "$__gc_root" = / ] || [ "${__gc_root%/}" = "$__gc_root" ] || return 1;
__gc_source_realpath "$__gc_root" || return 1; __gc_root=$__gc_resolved;
[ "$__gc_root" != / ] && [ "$__gc_candidate" != / ] || return 1;
case "$__gc_candidate" in "$__gc_root"|"$__gc_root"/*) return 1;; esac;
case "$__gc_root" in "$__gc_candidate"/*) return 1;; esac;
done < "$__gc_record";
done;
};
__gc_source_end() {
if [ -n "${__gc_reservation-}" ]; then
if [ -L "$__gc_reservation" ] || [ ! -f "$__gc_reservation" ]; then
exec 8>&-; return 1;
fi;
__gc_actual=$(sha256sum -- "$__gc_reservation") || { exec 8>&-; return 1; };
[ "${__gc_actual%% *}" = "$__gc_reservation_digest" ] || { exec 8>&-; return 1; };
[ ! -L "$__gc_source_registry/released" ] || { exec 8>&-; return 1; };
mkdir -p -- "$__gc_source_registry/released" || { exec 8>&-; return 1; };
__gc_receipt="$__gc_source_registry/released/${__gc_reservation##*/}";
[ ! -e "$__gc_receipt" ] && [ ! -L "$__gc_receipt" ] || { exec 8>&-; return 1; };
if ! mv -- "$__gc_reservation" "$__gc_receipt" || ! sync -f "$__gc_receipt" \
|| ! sync -f "$__gc_source_registry/released" || ! sync -f "$__gc_source_registry"; then
exec 8>&-; return 1;
fi;
__gc_reservation="";
fi;
exec 8>&-;
};
__gc_source_begin() {
__gc_reservation="";
[ ! -L "$__gc_source_registry" ] || return 1;
(umask 077; mkdir -p -- "$__gc_source_registry") || return 1;
[ ! -L "$__gc_source_registry/metadata.lock" ] || return 1;
exec 8>"$__gc_source_registry/metadata.lock" || return 1;
if ! flock -x 8; then __gc_source_end; return 1; fi;
if [ ! -d "$1" ] || [ -L "$1" ] || ! __gc_source_records_free "$1"; then
__gc_source_end; return 1;
fi;
__gc_token=fc-$(cat /proc/sys/kernel/random/uuid) || { __gc_source_end; return 1; };
case "$__gc_token" in *[!a-f0-9-]*) __gc_source_end; return 1;; esac;
[ "${#__gc_token}" -eq 39 ] || { __gc_source_end; return 1; };
for __gc_previous in "$__gc_source_registry/$__gc_token."* \
"$__gc_source_registry/released/$__gc_token."* "$__gc_source_registry/cancelled/$__gc_token."*; do
if [ -e "$__gc_previous" ] || [ -L "$__gc_previous" ]; then __gc_source_end; return 1; fi;
done;
__gc_actual=$(printf '%s\n' "$__gc_candidate" | sha256sum) || { __gc_source_end; return 1; };
__gc_reservation_digest=${__gc_actual%% *};
__gc_pending="$__gc_source_registry/$__gc_token.$__gc_reservation_digest.pending";
if ! (umask 077; set -C; printf '%s\n' "$__gc_candidate" > "$__gc_pending") \
|| ! sync -f "$__gc_pending"; then __gc_source_end; return 1; fi;
__gc_active="$__gc_source_registry/$__gc_token.$__gc_reservation_digest.claim";
if ! mv -- "$__gc_pending" "$__gc_active" || ! sync -f "$__gc_source_registry"; then
__gc_source_end; return 1;
fi;
__gc_reservation=$__gc_active;
};
"#;
format!(
"__gc_source_registry={}; {GATE}",
shell_escape::escape(registry.into())
)
}
const TMPBASE_MAXDEPTH: u32 = 6;
fn dedup_candidate_file(var: &str) -> String {
format!(
"if __ded=$(mktemp 2>/dev/null || mktemp -p \"$__tmpbase\" 2>/dev/null); then \
if sort -u \"${var}\" -o \"$__ded\" 2>/dev/null; then mv -f \"$__ded\" \"${var}\"; \
else rm -f \"$__ded\"; fi; \
fi; "
)
}
fn gate_snapshot_fragment() -> String {
concat!(
"__held=\"\"; __held_ok=0; __hsrc=none; __proc=\"\"; __proc_ok=0; ",
"if __held=$(mktemp 2>/dev/null || mktemp -p \"$__tmpbase\" 2>/dev/null); then ",
" if [ -d /proc/self/fd ]; then ",
" find /proc/[0-9]*/cwd /proc/[0-9]*/fd -maxdepth 1 -printf '%l\\n' 2>/dev/null > \"$__held\" || :; ",
" fi; ",
" if [ -s \"$__held\" ]; then __hsrc=proc; ",
" elif command -v lsof >/dev/null 2>&1; then ",
" lsof -w -n -P -F n 2>/dev/null | sed -n 's/^n//p' > \"$__held\" || :; ",
" if [ -s \"$__held\" ]; then __hsrc=lsof; fi; ",
" fi; ",
" if [ -s \"$__held\" ]; then __held_ok=1; fi; ",
"fi; ",
"__self_pids=\" $$ \"; __pp=$$; __i=0; ",
"while [ \"$__i\" -lt 12 ]; do ",
" __pp=$(ps -o ppid= -p \"$__pp\" 2>/dev/null | tr -d ' '); ",
" case \"$__pp\" in ''|0|1) break;; *[!0-9]*) break;; esac; ",
" __self_pids=\"$__self_pids$__pp \"; __i=$((__i + 1)); ",
"done; ",
"if __proc=$(mktemp 2>/dev/null || mktemp -p \"$__tmpbase\" 2>/dev/null); then ",
" ps -ww -eo pid=,args= > \"$__proc\" 2>/dev/null ",
" || ps -eo pid=,args= > \"$__proc\" 2>/dev/null || :; ",
" if [ -s \"$__proc\" ]; then __proc_ok=1; fi; ",
"fi; ",
"__gc_handles() { ",
" if [ \"$__held_ok\" -ne 1 ]; then printf unknown; return 0; fi; ",
" if awk -v p=\"$1\" '$0==p || index($0, p \"/\")==1 {f=1; exit} END{exit(f?0:1)}' \"$__held\"; ",
" then printf held; else printf free; fi; ",
"}; ",
"__gc_procs() { ",
" if [ \"$__proc_ok\" -ne 1 ]; then printf unknown; return 0; fi; ",
" if awk -v p=\"$1\" -v skip=\"$__self_pids\" ",
" '{ if (index(skip, \" \" $1 \" \") > 0) next; if (index($0, p) > 0) { f=1; exit } } ",
" END{exit(f?0:1)}' \"$__proc\"; ",
" then printf held; else printf free; fi; ",
"}; ",
"__gc_gates_ok() { ",
" [ \"$(__gc_handles \"$1\")\" = free ] || return 1; ",
" [ \"$(__gc_procs \"$1\")\" = free ] || return 1; ",
" return 0; ",
"}; ",
"printf 'RCH_GC_GATES handles=%s handles_source=%s procs=%s\\n' \"$__held_ok\" \"$__hsrc\" \"$__proc_ok\"; ",
)
.to_string()
}
fn gate_snapshot_cleanup() -> &'static str {
"if [ -n \"$__held\" ]; then rm -f \"$__held\"; fi; \
if [ -n \"$__proc\" ]; then rm -f \"$__proc\"; fi; "
}
fn pooled_collect_gate(idle_minutes: u64) -> String {
format!(
"[ -d \"$d\" ] || continue; \
if find \"$d\" -mmin -{idle_minutes} -print -quit 2>/dev/null | grep -q .; then continue; fi; \
if ! __gc_gates_ok \"$d\"; then printf 'RCH_GC_SKIP pooled-ttl gate %s\\n' \"$d\"; continue; fi; "
)
}
fn candidate_discovery_preamble(escaped_base: &str, on_guard_exit: &str) -> String {
let tmp_base_prelude = crate::gc_roots::remote_cargo_home_base_prelude();
let tmp_base_var = crate::gc_roots::RCH_CARGO_HOME_BASE_VAR;
format!(
"set -u; \
base=\"{escaped_base}\"; \
printf 'RCH_GC_ROOT base %s\\n' \"$base\"; \
if [ ! -d \"$base\" ]; then printf 'RCH_GC_ROOT_SKIPPED %s missing\\n' \"$base\"; {on_guard_exit}exit 0; fi; \
__rt=$(cd \"$base\" 2>/dev/null && pwd -P) || {{ printf 'RCH_GC_ROOT_SKIPPED %s unresolvable\\n' \"$base\"; {on_guard_exit}exit 0; }}; \
[ -n \"$__rt\" ] || {{ printf 'RCH_GC_ROOT_SKIPPED %s unresolvable\\n' \"$base\"; {on_guard_exit}exit 0; }}; \
case \"$__rt\" in */*/*) ;; *) printf 'RCH_GC_ROOT_SKIPPED %s too-shallow\\n' \"$__rt\"; {on_guard_exit}exit 0;; esac; \
printf 'RCH_GC_ROOT resolved %s\\n' \"$__rt\"; \
{tmp_base_prelude}; \
__tmpbase=\"${{{tmp_base_var}}}\"; \
__tmpscan=\"\"; \
case \"$__tmpbase\" in /*/*) __tmpscan=$(cd \"$__tmpbase\" 2>/dev/null && pwd -P) || __tmpscan=\"\";; esac; \
case \"$__tmpscan\" in /*/*) ;; *) __tmpscan=\"\";; esac; \
printf 'RCH_GC_ROOT tmp %s\\n' \"$__tmpscan\"; \
{gates}\
__tmpf=$(mktemp 2>/dev/null || mktemp -p \"$__tmpbase\" 2>/dev/null) || {{ {cleanup}{on_guard_exit}exit 0; }}; \
find \"$__rt\" -maxdepth 8 -type d \\( -name \".rch-target-*-job-*\" -o -name \".rch-target-*-pid-*\" \\) -prune 2>/dev/null > \"$__tmpf\"; \
if [ -n \"$__tmpscan\" ]; then find \"$__tmpscan\" -maxdepth 1 -type d -name \"rch_target_*\" -prune 2>/dev/null >> \"$__tmpf\"; \
find \"$__tmpscan\" -maxdepth {TMPBASE_MAXDEPTH} -type d \\( -name \".rch-target-*-job-*\" -o -name \".rch-target-*-pid-*\" \\) -prune 2>/dev/null >> \"$__tmpf\"; fi; \
{dedup}",
gates = gate_snapshot_fragment(),
cleanup = gate_snapshot_cleanup(),
dedup = dedup_candidate_file("__tmpf")
)
}
pub const MIN_POOLED_IDLE_MINUTES: u64 = 24 * 60;
#[must_use]
pub fn worker_sweep_command(
escaped_base: &str,
idle_minutes: u64,
pooled_idle_minutes: Option<u64>,
max_cache_kb: Option<u64>,
) -> String {
worker_sweep_command_with_registry(
escaped_base,
idle_minutes,
pooled_idle_minutes,
max_cache_kb,
SOURCE_CLAIM_REGISTRY,
)
}
#[must_use]
pub fn worker_sweep_command_with_registry(
escaped_base: &str,
idle_minutes: u64,
pooled_idle_minutes: Option<u64>,
max_cache_kb: Option<u64>,
registry: &str,
) -> String {
let loop_body = reap_loop_body_with_source_gate(
idle_minutes,
None,
"removed",
"freed_kb",
Some("ttl"),
true,
);
let guard = "printf 'RCH_WORKER_REAP_METRICS removed=0 freed_kb=0\\n'; ";
let preamble = candidate_discovery_preamble(escaped_base, guard);
let pooled_pass = match pooled_idle_minutes {
Some(window) => {
let window = window.max(MIN_POOLED_IDLE_MINUTES);
let pooled_body = reap_loop_body_with_source_gate(
window,
None,
"removed",
"freed_kb",
Some("pooled-ttl"),
true,
);
let dedup2 = dedup_candidate_file("__tmpf2");
format!(
"if __tmpf2=$(mktemp 2>/dev/null || mktemp -p \"$__tmpbase\" 2>/dev/null); then \
find \"$__rt\" -maxdepth 8 -type d -name \".rch-target-*-pool-*\" -prune 2>/dev/null > \"$__tmpf2\"; \
if [ -n \"$__tmpscan\" ]; then find \"$__tmpscan\" -maxdepth {TMPBASE_MAXDEPTH} -type d -name \".rch-target-*-pool-*\" -prune 2>/dev/null >> \"$__tmpf2\"; fi; \
{dedup2}\
while IFS= read -r d; do {pooled_gate}{pooled_body} done < \"$__tmpf2\"; \
rm -f \"$__tmpf2\"; \
fi; ",
pooled_gate = pooled_collect_gate(window),
)
}
None => String::new(),
};
let dedup3 = dedup_candidate_file("__tmpf3");
let cap_pass = match max_cache_kb {
Some(cap_kb) => format!(
"if [ {cap_kb} -gt 0 ] \
&& __lst=$(mktemp 2>/dev/null || mktemp -p \"$__tmpbase\" 2>/dev/null) \
&& __tmpf3=$(mktemp 2>/dev/null || mktemp -p \"$__tmpbase\" 2>/dev/null); then \
find \"$__rt\" -maxdepth 8 -type d \\( -name \".rch-target-*-job-*\" -o -name \".rch-target-*-pid-*\" -o -name \".rch-target-*-pool-*\" \\) -prune 2>/dev/null > \"$__tmpf3\"; \
if [ -n \"$__tmpscan\" ]; then find \"$__tmpscan\" -maxdepth 1 -type d -name \"rch_target_*\" -prune 2>/dev/null >> \"$__tmpf3\"; \
find \"$__tmpscan\" -maxdepth {TMPBASE_MAXDEPTH} -type d \\( -name \".rch-target-*-job-*\" -o -name \".rch-target-*-pid-*\" -o -name \".rch-target-*-pool-*\" \\) -prune 2>/dev/null >> \"$__tmpf3\"; fi; \
{dedup3}\
: > \"$__lst\"; total_kb=0; \
while IFS= read -r d; do \
[ -d \"$d\" ] || continue; \
__n=$(date -r \"$d\" +%s 2>/dev/null); [ -n \"$__n\" ] || __n=0; \
__k=$(du -sk \"$d\" 2>/dev/null | awk '{{print $1}}'); [ -n \"$__k\" ] || __k=0; \
total_kb=$((total_kb + __k)); \
printf '%s %s %s\\n' \"$__n\" \"$__k\" \"$d\" >> \"$__lst\"; \
done < \"$__tmpf3\"; \
if [ \"$total_kb\" -gt {cap_kb} ]; then \
printf 'RCH_REAP_CAP initial_kb=%s cap_kb={cap_kb}\\n' \"$total_kb\"; \
sort -n \"$__lst\" > \"$__tmpf3\"; \
while IFS=' ' read -r __n __k d; do \
[ \"$total_kb\" -le {cap_kb} ] && break; \
[ -d \"$d\" ] || continue; \
if find \"$d\" -mmin -{idle_minutes} -print -quit 2>/dev/null | grep -q .; then \
printf 'RCH_GC_SKIP cap active %s\\n' \"$d\"; continue; \
fi; \
if ! __gc_gates_ok \"$d\"; then printf 'RCH_GC_SKIP cap gate %s\\n' \"$d\"; continue; fi; \
if ! __gc_source_begin \"$d\"; then printf 'RCH_GC_SKIP cap source-ownership %s\\n' \"$d\"; continue; fi; \
if find \"$d\" -mmin -{idle_minutes} -print -quit 2>/dev/null | grep -q .; then \
__gc_source_end; printf 'RCH_GC_SKIP cap active %s\\n' \"$d\"; continue; \
fi; \
if __rmerr=$(rm -rf -- \"$d\" 2>&1); then \
removed=$((removed + 1)); freed_kb=$((freed_kb + __k)); total_kb=$((total_kb - __k)); \
printf 'RCH_REAP_RM %s cap %s\\n' \"$__k\" \"$d\"; \
else \
printf 'RCH_REAP_ERR cap %s :: %s\\n' \"$d\" \"$(printf '%s' \"$__rmerr\" | head -1)\"; \
fi; \
__gc_source_end; \
done < \"$__tmpf3\"; \
printf 'RCH_REAP_CAP final_kb=%s\\n' \"$total_kb\"; \
fi; \
rm -f \"$__lst\" \"$__tmpf3\"; \
fi; "
),
None => String::new(),
};
format!(
"{preamble}\
{source_gate}\
removed=0; freed_kb=0; \
while IFS= read -r d; do {loop_body} done < \"$__tmpf\"; \
rm -f \"$__tmpf\"; \
{pooled_pass}\
{cap_pass}\
{gate_cleanup}\
printf 'RCH_WORKER_REAP_METRICS removed=%s freed_kb=%s\\n' \"$removed\" \"$freed_kb\"",
gate_cleanup = gate_snapshot_cleanup(),
source_gate = source_claim_gate_fragment(registry),
)
}
#[must_use]
pub fn parse_worker_reap_metrics(stdout: &str) -> Option<(u64, u64)> {
let line = stdout
.lines()
.find(|l| l.contains("RCH_WORKER_REAP_METRICS"))?;
let mut removed = None;
let mut freed_kb = None;
for token in line.split_whitespace().skip(1) {
let Some((key, value)) = token.split_once('=') else {
continue;
};
match key {
"removed" => removed = value.parse::<u64>().ok(),
"freed_kb" => freed_kb = value.parse::<u64>().ok(),
_ => {}
}
}
Some((removed.unwrap_or(0), freed_kb.unwrap_or(0)))
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ReapEvent {
pub kb: u64,
pub trigger: String,
pub path: String,
}
#[must_use]
pub fn parse_reap_events(stdout: &str) -> Vec<ReapEvent> {
stdout
.lines()
.filter_map(|line| {
let rest = line.trim().strip_prefix("RCH_REAP_RM ")?;
let mut parts = rest.splitn(3, ' ');
let kb = parts.next()?.parse::<u64>().ok()?;
let trigger = parts.next()?.to_string();
let path = parts.next()?.trim();
if path.is_empty() {
return None;
}
Some(ReapEvent {
kb,
trigger,
path: path.to_string(),
})
})
.collect()
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ReapError {
pub trigger: String,
pub path: String,
pub message: String,
}
#[must_use]
pub fn parse_reap_errors(stdout: &str) -> Vec<ReapError> {
stdout
.lines()
.filter_map(|line| {
let rest = line.trim().strip_prefix("RCH_REAP_ERR ")?;
let (trigger, rest) = rest.split_once(' ')?;
let (path, message) = rest.split_once(" :: ")?;
if path.is_empty() {
return None;
}
Some(ReapError {
trigger: trigger.to_string(),
path: path.to_string(),
message: message.trim().to_string(),
})
})
.collect()
}
#[must_use]
pub fn enumerate_targets_command(escaped_base: &str) -> String {
let preamble = candidate_discovery_preamble(escaped_base, "");
let dedup = dedup_candidate_file("__tmpf");
format!(
"{preamble}\
find \"$__rt\" -maxdepth 8 -type d \\( -name \".rch-target-*-pool-*\" -o -name \"{cache_glob}\" \\) -prune 2>/dev/null >> \"$__tmpf\"; \
if [ -n \"$__tmpscan\" ]; then \
find \"$__tmpscan\" -maxdepth {TMPBASE_MAXDEPTH} -type d \\( -name \".rch-target-*-pool-*\" -o -name \"{cache_glob}\" \\) -prune 2>/dev/null >> \"$__tmpf\"; fi; \
{dedup}\
while IFS= read -r d; do \
[ -d \"$d\" ] || continue; \
set -- $(find \"$d\" -printf '%T@ %s\\n' 2>/dev/null | awk '{{ t=int($1); if (t>n) n=t; s+=$2 }} END {{ printf \"%d %d\", n, int(s/1024) }}'); \
newest=${{1:-0}}; kb=${{2:-0}}; \
printf 'RCH_TARGET_ENTRY %s %s %s %s %s\\n' \"$newest\" \"$kb\" \"$(__gc_handles \"$d\")\" \"$(__gc_procs \"$d\")\" \"$d\"; \
done < \"$__tmpf\"; \
rm -f \"$__tmpf\"; \
{gate_cleanup}exit 0",
cache_glob = CARGO_CACHE_GLOB,
gate_cleanup = gate_snapshot_cleanup()
)
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RemoteTargetEntry {
pub newest_mtime_unix: u64,
pub kb: u64,
pub path: String,
pub open_handles: GateEvidence,
pub live_process: GateEvidence,
}
impl RemoteTargetEntry {
#[must_use]
pub fn is_pooled(&self) -> bool {
self.class() == GcClass::Pooled
}
#[must_use]
pub fn class(&self) -> GcClass {
GcClass::from_path(&self.path)
}
#[must_use]
pub fn age_secs(&self, now_unix: u64) -> u64 {
now_unix.saturating_sub(self.newest_mtime_unix)
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub enum GcClass {
PerJob,
Pooled,
CargoCache,
LegacyTarget,
Unrecognized,
}
impl GcClass {
#[must_use]
pub fn from_path(path: &str) -> Self {
let name = path.trim_end_matches('/').rsplit('/').next().unwrap_or("");
if let Some(rest) = name.strip_prefix(".rch-target-") {
if rest.contains("-pool-") {
return Self::Pooled;
}
if rest.contains("-job-") || rest.contains("-pid-") {
return Self::PerJob;
}
return Self::Unrecognized;
}
if name.starts_with(crate::gc_roots::RCH_CARGO_CACHE_PREFIX) {
return Self::CargoCache;
}
if name.starts_with("rch_target_") {
return Self::LegacyTarget;
}
Self::Unrecognized
}
#[must_use]
pub const fn as_str(self) -> &'static str {
match self {
Self::PerJob => "per_job",
Self::Pooled => "pooled",
Self::CargoCache => "cargo_cache",
Self::LegacyTarget => "legacy_target",
Self::Unrecognized => "unrecognized",
}
}
#[must_use]
pub const fn trigger(self) -> &'static str {
match self {
Self::PerJob | Self::LegacyTarget => "ttl",
Self::Pooled => "pooled-ttl",
Self::CargoCache => "cache-ttl",
Self::Unrecognized => "none",
}
}
#[must_use]
pub const fn requires_known_gates(self) -> bool {
matches!(self, Self::Pooled | Self::CargoCache)
}
}
impl std::fmt::Display for GcClass {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str(self.as_str())
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Default)]
pub enum GateEvidence {
Free,
Held,
#[default]
Unknown,
}
impl GateEvidence {
#[must_use]
pub fn from_token(token: &str) -> Option<Self> {
match token {
"free" => Some(Self::Free),
"held" => Some(Self::Held),
"unknown" => Some(Self::Unknown),
_ => None,
}
}
#[must_use]
pub const fn as_token(self) -> &'static str {
match self {
Self::Free => "free",
Self::Held => "held",
Self::Unknown => "unknown",
}
}
}
impl std::fmt::Display for GateEvidence {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str(self.as_token())
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct GcPolicy {
pub idle_secs: u64,
pub pooled_idle_secs: Option<u64>,
pub cache_idle_secs: Option<u64>,
}
impl GcPolicy {
#[must_use]
pub fn window_secs(&self, class: GcClass) -> Option<u64> {
match class {
GcClass::PerJob | GcClass::LegacyTarget => Some(self.idle_secs),
GcClass::Pooled => self.pooled_idle_secs,
GcClass::CargoCache => self.cache_idle_secs,
GcClass::Unrecognized => None,
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum GcKeepReason {
Unrecognized,
ClassDisabled,
TooYoung {
age_secs: u64,
required_secs: u64,
},
OpenHandles,
LiveProcess,
GateUnavailable(&'static str),
}
impl std::fmt::Display for GcKeepReason {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::Unrecognized => f.write_str("not an rch-managed runtime dir"),
Self::ClassDisabled => f.write_str("collection disabled for this class by config"),
Self::TooYoung {
age_secs,
required_secs,
} => write!(
f,
"idle {}h < required {}h",
age_secs / 3600,
required_secs / 3600
),
Self::OpenHandles => f.write_str("open file descriptors under the dir"),
Self::LiveProcess => f.write_str("a live process is rooted at the dir"),
Self::GateUnavailable(gate) => write!(f, "{gate} gate unavailable (treated as in use)"),
}
}
}
impl GcKeepReason {
#[must_use]
pub const fn as_str(&self) -> &'static str {
match self {
Self::Unrecognized => "unrecognized",
Self::ClassDisabled => "class_disabled",
Self::TooYoung { .. } => "too_young",
Self::OpenHandles => "open_handles",
Self::LiveProcess => "live_process",
Self::GateUnavailable(_) => "gate_unavailable",
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum GcVerdict {
Collect,
Keep(GcKeepReason),
}
impl GcVerdict {
#[must_use]
pub fn is_collect(&self) -> bool {
matches!(self, Self::Collect)
}
}
#[must_use]
pub fn evaluate_gc_candidate(
entry: &RemoteTargetEntry,
now_unix: u64,
policy: &GcPolicy,
) -> GcVerdict {
let class = entry.class();
if class == GcClass::Unrecognized {
return GcVerdict::Keep(GcKeepReason::Unrecognized);
}
let Some(required_secs) = policy.window_secs(class) else {
return GcVerdict::Keep(GcKeepReason::ClassDisabled);
};
let age_secs = entry.age_secs(now_unix);
if age_secs < required_secs {
return GcVerdict::Keep(GcKeepReason::TooYoung {
age_secs,
required_secs,
});
}
let strict = class.requires_known_gates();
match entry.open_handles {
GateEvidence::Held => return GcVerdict::Keep(GcKeepReason::OpenHandles),
GateEvidence::Unknown if strict => {
return GcVerdict::Keep(GcKeepReason::GateUnavailable("open-descriptor"));
}
_ => {}
}
match entry.live_process {
GateEvidence::Held => return GcVerdict::Keep(GcKeepReason::LiveProcess),
GateEvidence::Unknown if strict => {
return GcVerdict::Keep(GcKeepReason::GateUnavailable("live-process"));
}
_ => {}
}
GcVerdict::Collect
}
#[must_use]
pub fn select_cap_evictions(
entries: &[RemoteTargetEntry],
verdicts: &[GcVerdict],
now_unix: u64,
policy: &GcPolicy,
cap_kb: u64,
) -> Vec<usize> {
if cap_kb == 0 || entries.len() != verdicts.len() {
return Vec::new();
}
let mut remaining_kb: u64 = entries
.iter()
.zip(verdicts)
.filter(|(_, v)| !v.is_collect())
.map(|(e, _)| e.kb)
.sum();
if remaining_kb <= cap_kb {
return Vec::new();
}
let mut candidates: Vec<usize> = entries
.iter()
.zip(verdicts)
.enumerate()
.filter(|(_, (entry, verdict))| {
if verdict.is_collect() {
return false;
}
let class = entry.class();
if class == GcClass::Unrecognized || policy.window_secs(class).is_none() {
return false;
}
if entry.age_secs(now_unix) < policy.idle_secs {
return false;
}
let strict = class.requires_known_gates();
let gate_ok = |g: GateEvidence| match g {
GateEvidence::Free => true,
GateEvidence::Held => false,
GateEvidence::Unknown => !strict,
};
gate_ok(entry.open_handles) && gate_ok(entry.live_process)
})
.map(|(idx, _)| idx)
.collect();
candidates.sort_by(|a, b| {
entries[*a]
.newest_mtime_unix
.cmp(&entries[*b].newest_mtime_unix)
.then_with(|| entries[*a].path.cmp(&entries[*b].path))
});
let mut evict = Vec::new();
for idx in candidates {
if remaining_kb <= cap_kb {
break;
}
remaining_kb = remaining_kb.saturating_sub(entries[idx].kb);
evict.push(idx);
}
evict
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct GcCollectTarget {
pub path: String,
pub idle_minutes: u64,
pub trigger: &'static str,
}
pub fn collect_paths_command(targets: &[GcCollectTarget]) -> Result<String, String> {
collect_paths_command_with_registry(targets, SOURCE_CLAIM_REGISTRY)
}
fn collect_paths_command_with_registry(
targets: &[GcCollectTarget],
registry: &str,
) -> Result<String, String> {
if targets.is_empty() {
return Err("no targets to collect".to_string());
}
let mut list = String::new();
for target in targets {
if !is_safe_reap_path(&target.path) {
return Err(format!(
"refusing to collect {:?}: not an absolute, `..`-free, metacharacter-free path at \
least two levels deep",
target.path
));
}
if GcClass::from_path(&target.path) == GcClass::Unrecognized {
return Err(format!(
"refusing to collect {:?}: basename is not an rch-managed runtime dir",
target.path
));
}
if target.idle_minutes == 0 {
return Err(format!(
"refusing to collect {:?}: a zero idle window would remove a live dir",
target.path
));
}
if !target
.trigger
.chars()
.all(|c| c.is_ascii_alphanumeric() || c == '-')
{
return Err(format!("invalid trigger tag {:?}", target.trigger));
}
list.push_str(&format!(
" \"{}:{}:{}\"",
target.idle_minutes, target.trigger, target.path
));
}
let tmp_base_prelude = crate::gc_roots::remote_cargo_home_base_prelude();
let tmp_base_var = crate::gc_roots::RCH_CARGO_HOME_BASE_VAR;
Ok(format!(
"set -u; \
{tmp_base_prelude}; \
__tmpbase=\"${{{tmp_base_var}}}\"; \
{gates}\
{source_gate}\
removed=0; freed_kb=0; \
for __e in{list}; do \
__mins=${{__e%%:*}}; __r=${{__e#*:}}; __tag=${{__r%%:*}}; d=${{__r#*:}}; \
if [ ! -d \"$d\" ]; then printf 'RCH_GC_SKIP %s missing %s\\n' \"$__tag\" \"$d\"; continue; fi; \
if find \"$d\" -mmin -\"$__mins\" -print -quit 2>/dev/null | grep -q .; then \
printf 'RCH_GC_SKIP %s active %s\\n' \"$__tag\" \"$d\"; continue; fi; \
if ! __gc_gates_ok \"$d\"; then printf 'RCH_GC_SKIP %s gate %s\\n' \"$__tag\" \"$d\"; continue; fi; \
if ! __gc_source_begin \"$d\"; then printf 'RCH_GC_SKIP %s source-ownership %s\\n' \"$__tag\" \"$d\"; continue; fi; \
if find \"$d\" -mmin -\"$__mins\" -print -quit 2>/dev/null | grep -q .; then \
__gc_source_end; printf 'RCH_GC_SKIP %s active %s\\n' \"$__tag\" \"$d\"; continue; fi; \
sz=$(du -sk \"$d\" 2>/dev/null | awk '{{print $1}}'); [ -z \"$sz\" ] && sz=0; \
if __rmerr=$(rm -rf -- \"$d\" 2>&1); then \
removed=$((removed + 1)); freed_kb=$((freed_kb + sz)); \
printf 'RCH_REAP_RM %s %s %s\\n' \"$sz\" \"$__tag\" \"$d\"; \
else \
printf 'RCH_REAP_ERR %s %s :: %s\\n' \"$__tag\" \"$d\" \"$(printf '%s' \"$__rmerr\" | head -1)\"; \
fi; \
__gc_source_end; \
done; \
{gate_cleanup}\
printf 'RCH_WORKER_REAP_METRICS removed=%s freed_kb=%s\\n' \"$removed\" \"$freed_kb\"",
gates = gate_snapshot_fragment(),
gate_cleanup = gate_snapshot_cleanup(),
source_gate = source_claim_gate_fragment(registry),
))
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct GcSkip {
pub trigger: String,
pub reason: String,
pub path: String,
}
#[must_use]
pub fn parse_gc_skips(stdout: &str) -> Vec<GcSkip> {
stdout
.lines()
.filter_map(|line| {
let rest = line.trim().strip_prefix("RCH_GC_SKIP ")?;
let mut parts = rest.splitn(3, ' ');
let trigger = parts.next()?.to_string();
let reason = parts.next()?.to_string();
let path = parts.next()?.trim();
if path.is_empty() {
return None;
}
Some(GcSkip {
trigger,
reason,
path: path.to_string(),
})
})
.collect()
}
pub const ABANDONED_GC_CLAIM_MIN_MINUTES: u64 = 6 * 60;
#[must_use]
pub fn abandoned_gc_claims_command(apply: bool) -> String {
abandoned_gc_claims_command_with_registry(
apply,
SOURCE_CLAIM_REGISTRY,
ABANDONED_GC_CLAIM_MIN_MINUTES,
)
}
#[must_use]
pub fn abandoned_gc_claims_command_with_registry(
apply: bool,
registry: &str,
min_age_minutes: u64,
) -> String {
let registry = shell_escape::escape(registry.into());
let apply = u8::from(apply);
format!(
r#"r={registry}; min={min_age_minutes}; apply={apply};
if [ ! -d "$r" ] || [ -L "$r" ]; then printf 'RCH_GC_CLAIMS none\n'; exit 0; fi;
if [ -L "$r/metadata.lock" ]; then printf 'RCH_GC_CLAIMS unsafe-lock\n'; exit 0; fi;
if ! command -v flock >/dev/null 2>&1; then printf 'RCH_GC_CLAIMS no-flock\n'; exit 0; fi;
exec 8>"$r/metadata.lock" || exit 1;
if ! flock -x -w 30 8; then printf 'RCH_GC_CLAIMS busy\n'; exit 0; fi;
if [ ! -d /proc/self/fd ]; then printf 'RCH_GC_CLAIMS no-proc\n'; exit 0; fi;
refs=$(mktemp) && args=$(mktemp) || exit 1;
trap 'rm -f -- "$refs" "$args"' EXIT;
find /proc/[0-9]*/cwd /proc/[0-9]*/root /proc/[0-9]*/fd -maxdepth 1 -printf '%l\n' 2>/dev/null | grep '^/' | sort -u > "$refs";
for p in /proc/[0-9]*; do tr '\0' ' ' 2>/dev/null < "$p/cmdline"; echo; done > "$args";
printf 'RCH_GC_CLAIMS ok\n';
now=$(date +%s);
for c in "$r"/fc-*.claim; do
[ -e "$c" ] || [ -L "$c" ] || continue;
n=${{c##*/}}; p=-; age=0; v=keep;
if [ -L "$c" ] || [ ! -f "$c" ]; then why=unsafe-record;
else
m=$(stat -c %Y -- "$c" 2>/dev/null) || m=$now; age=$(( (now - m) / 60 ));
p=$(head -n 1 -- "$c"); d=${{n#*.}}; d=${{d%%.*}}; a=$(sha256sum -- "$c");
if [ "${{a%% *}}" != "$d" ] || [ -z "$p" ]; then why=unsafe-record;
elif [ "$age" -lt "$min" ]; then why=young;
elif awk -v p="$p/" -v q="$p" 'index($0,p)==1 || $0==q {{f=1; exit}} END{{exit !f}}' "$refs"; then why=open-handle;
elif grep -qF -- "$p" "$args"; then why=live-process;
elif [ "$apply" = 1 ]; then
dest="$r/released/$n";
if [ ! -L "$r/released" ] && mkdir -p -- "$r/released" && [ ! -e "$dest" ] && [ ! -L "$dest" ] \
&& mv -- "$c" "$dest" && sync -f "$dest" && sync -f "$r/released"; then v=released; why=abandoned;
else why=release-failed; fi;
else v=would-release; why=abandoned; fi;
fi;
printf 'RCH_GC_CLAIM %s %s %s %s %s\n' "$v" "$why" "$age" "$n" "$p";
done;
sync -f "$r" 2>/dev/null || :;
"#
)
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct GcClaimReport {
pub verdict: String,
pub reason: String,
pub age_minutes: u64,
pub name: String,
pub path: String,
}
#[must_use]
pub fn parse_gc_claim_reports(stdout: &str) -> (Option<String>, Vec<GcClaimReport>) {
let mut status = None;
let mut claims = Vec::new();
for line in stdout.lines() {
let line = line.trim();
if let Some(rest) = line.strip_prefix("RCH_GC_CLAIMS ") {
status = Some(rest.to_string());
} else if let Some(rest) = line.strip_prefix("RCH_GC_CLAIM ") {
let mut parts = rest.splitn(5, ' ');
let (Some(verdict), Some(reason), Some(age), Some(name), Some(path)) = (
parts.next(),
parts.next(),
parts.next(),
parts.next(),
parts.next(),
) else {
continue;
};
let Ok(age_minutes) = age.parse() else {
continue;
};
claims.push(GcClaimReport {
verdict: verdict.to_string(),
reason: reason.to_string(),
age_minutes,
name: name.to_string(),
path: path.to_string(),
});
}
}
(status, claims)
}
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct ScanRootReport {
pub requested: Option<String>,
pub resolved: Option<String>,
pub temp_base: Option<String>,
pub skipped: Option<String>,
}
#[must_use]
pub fn parse_scan_roots(stdout: &str) -> ScanRootReport {
let mut report = ScanRootReport::default();
for line in stdout.lines() {
let line = line.trim();
if let Some(rest) = line.strip_prefix("RCH_GC_ROOT_SKIPPED ") {
if !rest.is_empty() {
report.skipped = Some(rest.to_string());
}
continue;
}
let Some(rest) = line.strip_prefix("RCH_GC_ROOT ") else {
continue;
};
let Some((kind, value)) = rest.split_once(' ') else {
continue;
};
let value = value.trim();
let slot = match kind {
"base" => &mut report.requested,
"resolved" => &mut report.resolved,
"tmp" => &mut report.temp_base,
_ => continue,
};
*slot = (!value.is_empty()).then(|| value.to_string());
}
report
}
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct GateAvailability {
pub handles: bool,
pub processes: bool,
}
impl GateAvailability {
#[must_use]
pub fn both(self) -> bool {
self.handles && self.processes
}
}
#[must_use]
pub fn parse_gate_availability(stdout: &str) -> GateAvailability {
let mut out = GateAvailability::default();
let Some(line) = stdout.lines().find(|l| l.contains("RCH_GC_GATES")) else {
return out;
};
for token in line.split_whitespace() {
match token.split_once('=') {
Some(("handles", v)) => out.handles = v == "1",
Some(("procs", v)) => out.processes = v == "1",
_ => {}
}
}
out
}
#[must_use]
pub fn parse_target_entries(stdout: &str) -> Vec<RemoteTargetEntry> {
stdout
.lines()
.filter_map(|line| {
let rest = line.trim().strip_prefix("RCH_TARGET_ENTRY ")?;
let mut parts = rest.splitn(5, ' ');
let newest_mtime_unix = parts.next()?.parse::<u64>().ok()?;
let kb = parts.next()?.parse::<u64>().ok()?;
let third = parts.next()?;
let (open_handles, live_process, path) = match GateEvidence::from_token(third) {
Some(handles) => {
let live = GateEvidence::from_token(parts.next()?)?;
(handles, live, parts.next()?.trim().to_string())
}
None => (
GateEvidence::Unknown,
GateEvidence::Unknown,
rest.splitn(3, ' ').nth(2)?.trim().to_string(),
),
};
if path.is_empty() {
return None;
}
Some(RemoteTargetEntry {
newest_mtime_unix,
kb,
path,
open_handles,
live_process,
})
})
.collect()
}
#[cfg(test)]
mod tests {
use super::*;
fn entry(path: &str, age_secs: u64, gates: (GateEvidence, GateEvidence)) -> RemoteTargetEntry {
RemoteTargetEntry {
newest_mtime_unix: NOW - age_secs,
kb: 4_096,
path: path.to_string(),
open_handles: gates.0,
live_process: gates.1,
}
}
const NOW: u64 = 1_800_000_000;
const FREE: (GateEvidence, GateEvidence) = (GateEvidence::Free, GateEvidence::Free);
const DAY: u64 = 24 * 3600;
fn policy() -> GcPolicy {
GcPolicy {
idle_secs: 12 * 3600,
pooled_idle_secs: Some(7 * DAY),
cache_idle_secs: Some(14 * DAY),
}
}
const POOL: &str = "/data/projects/repo/.rch-target-hz2-pool-deadbeef";
const CACHE: &str = "/data/tmp/rch-cargo-cache-hz2";
#[test]
fn cargo_cache_glob_tracks_the_creating_prefix() {
assert_eq!(
CARGO_CACHE_GLOB,
format!("{}*", crate::gc_roots::RCH_CARGO_CACHE_PREFIX)
);
let expr = crate::gc_roots::remote_cargo_cache_expr("hz2");
assert!(expr.ends_with("/rch-cargo-cache-hz2"), "{expr}");
assert_eq!(
GcClass::from_path(&expr.replace("${RCH_CH_BASE}", "/data/tmp")),
GcClass::CargoCache
);
}
#[test]
fn gc_class_is_derived_from_the_basename() {
assert_eq!(GcClass::from_path(POOL), GcClass::Pooled);
assert_eq!(GcClass::from_path(CACHE), GcClass::CargoCache);
assert_eq!(
GcClass::from_path("/data/projects/r/.rch-target-hz2-job-1-2-0"),
GcClass::PerJob
);
assert_eq!(
GcClass::from_path("/data/projects/r/.rch-target-hz2-pid-42-1-0"),
GcClass::PerJob
);
assert_eq!(
GcClass::from_path("/data/tmp/rch_target_old"),
GcClass::LegacyTarget
);
for other in [
"/data/projects/repo/target",
"/data/projects/repo",
"/data/projects/repo/.git",
"/data/projects/repo/.rch-target-hz2",
"/data/tmp/rch-cargo-home-hz2-run-7",
] {
assert_eq!(
GcClass::from_path(other),
GcClass::Unrecognized,
"{other} must never be collectible"
);
}
}
#[test]
fn pool_and_cache_dirs_are_eligible_when_every_gate_passes() {
for (path, age) in [(POOL, 8 * DAY), (CACHE, 15 * DAY)] {
assert_eq!(
evaluate_gc_candidate(&entry(path, age, FREE), NOW, &policy()),
GcVerdict::Collect,
"{path} should be collectible"
);
}
}
#[test]
fn every_gate_independently_blocks_collection() {
let p = policy();
assert_eq!(
evaluate_gc_candidate(&entry(POOL, 6 * DAY, FREE), NOW, &p),
GcVerdict::Keep(GcKeepReason::TooYoung {
age_secs: 6 * DAY,
required_secs: 7 * DAY
})
);
assert_eq!(
evaluate_gc_candidate(&entry(CACHE, 13 * DAY, FREE), NOW, &p),
GcVerdict::Keep(GcKeepReason::TooYoung {
age_secs: 13 * DAY,
required_secs: 14 * DAY
})
);
assert_eq!(
evaluate_gc_candidate(
&entry(POOL, 99 * DAY, (GateEvidence::Held, GateEvidence::Free)),
NOW,
&p
),
GcVerdict::Keep(GcKeepReason::OpenHandles)
);
assert_eq!(
evaluate_gc_candidate(
&entry(POOL, 99 * DAY, (GateEvidence::Free, GateEvidence::Held)),
NOW,
&p
),
GcVerdict::Keep(GcKeepReason::LiveProcess)
);
let disabled = GcPolicy {
pooled_idle_secs: None,
cache_idle_secs: None,
..p
};
for path in [POOL, CACHE] {
assert_eq!(
evaluate_gc_candidate(&entry(path, 99 * DAY, FREE), NOW, &disabled),
GcVerdict::Keep(GcKeepReason::ClassDisabled)
);
}
assert_eq!(
evaluate_gc_candidate(
&entry("/data/projects/repo/target", 99 * DAY, FREE),
NOW,
&p
),
GcVerdict::Keep(GcKeepReason::Unrecognized)
);
}
#[test]
fn an_unavailable_gate_makes_a_warm_cache_dir_ineligible() {
let p = policy();
for gates in [
(GateEvidence::Unknown, GateEvidence::Free),
(GateEvidence::Free, GateEvidence::Unknown),
(GateEvidence::Unknown, GateEvidence::Unknown),
] {
for path in [POOL, CACHE] {
let verdict = evaluate_gc_candidate(&entry(path, 99 * DAY, gates), NOW, &p);
assert!(
matches!(verdict, GcVerdict::Keep(GcKeepReason::GateUnavailable(_))),
"{path} with gates {gates:?} must be ineligible, got {verdict:?}"
);
}
}
}
#[test]
fn per_job_dirs_keep_idle_window_authority_but_still_honour_a_live_gate() {
let p = policy();
let job = "/data/projects/repo/.rch-target-hz2-job-1-2-0";
assert_eq!(
evaluate_gc_candidate(
&entry(job, 2 * DAY, (GateEvidence::Unknown, GateEvidence::Unknown)),
NOW,
&p
),
GcVerdict::Collect
);
assert_eq!(
evaluate_gc_candidate(
&entry(job, 2 * DAY, (GateEvidence::Held, GateEvidence::Unknown)),
NOW,
&p
),
GcVerdict::Keep(GcKeepReason::OpenHandles)
);
}
#[test]
fn a_future_mtime_reads_as_zero_age() {
let mut e = entry(POOL, 0, FREE);
e.newest_mtime_unix = NOW + 10_000;
assert_eq!(e.age_secs(NOW), 0);
assert!(matches!(
evaluate_gc_candidate(&e, NOW, &policy()),
GcVerdict::Keep(GcKeepReason::TooYoung { .. })
));
}
#[test]
fn idle_window_helpers_floor_and_disable() {
assert_eq!(pooled_idle_minutes_from_hours(0), None);
assert_eq!(
pooled_idle_minutes_from_hours(1),
Some(MIN_POOLED_IDLE_MINUTES)
);
assert_eq!(pooled_idle_minutes_from_hours(168), Some(168 * 60));
assert_eq!(cargo_cache_idle_minutes_from_days(0), None);
assert_eq!(
cargo_cache_idle_minutes_from_days(1),
Some(MIN_CARGO_CACHE_IDLE_MINUTES)
);
assert_eq!(cargo_cache_idle_minutes_from_days(14), Some(14 * 24 * 60));
}
#[test]
fn legacy_entry_lines_parse_with_unknown_gates_and_are_ineligible() {
let legacy = format!("RCH_TARGET_ENTRY {} 4096 {POOL}\n", NOW - 99 * DAY);
let entries = parse_target_entries(&legacy);
assert_eq!(entries.len(), 1);
assert_eq!(entries[0].path, POOL);
assert_eq!(entries[0].open_handles, GateEvidence::Unknown);
assert_eq!(entries[0].live_process, GateEvidence::Unknown);
assert!(matches!(
evaluate_gc_candidate(&entries[0], NOW, &policy()),
GcVerdict::Keep(GcKeepReason::GateUnavailable(_))
));
let current = format!(
"RCH_TARGET_ENTRY {} 4096 free free {POOL}\n",
NOW - 99 * DAY
);
let entries = parse_target_entries(¤t);
assert_eq!(entries[0].path, POOL);
assert_eq!(entries[0].open_handles, GateEvidence::Free);
assert_eq!(
evaluate_gc_candidate(&entries[0], NOW, &policy()),
GcVerdict::Collect
);
}
#[test]
fn gate_availability_defaults_to_unavailable() {
assert_eq!(
parse_gate_availability("nothing here"),
GateAvailability::default()
);
assert!(!parse_gate_availability("").both());
let ok = parse_gate_availability("RCH_GC_GATES handles=1 handles_source=proc procs=1\n");
assert!(ok.both());
let partial =
parse_gate_availability("RCH_GC_GATES handles=1 handles_source=proc procs=0\n");
assert!(!partial.both());
assert!(partial.handles);
}
#[test]
fn scan_roots_and_skips_are_reported_back() {
let out = "RCH_GC_ROOT base /srv\nRCH_GC_ROOT_SKIPPED /srv too-shallow\n";
let roots = parse_scan_roots(out);
assert_eq!(roots.requested.as_deref(), Some("/srv"));
assert_eq!(roots.resolved, None);
assert_eq!(roots.skipped.as_deref(), Some("/srv too-shallow"));
let out = "RCH_GC_ROOT base /data/projects\nRCH_GC_ROOT resolved /data/projects\nRCH_GC_ROOT tmp \n";
let roots = parse_scan_roots(out);
assert_eq!(roots.resolved.as_deref(), Some("/data/projects"));
assert_eq!(
roots.temp_base, None,
"an unscanned tmp base is absent, not empty-string"
);
assert_eq!(roots.skipped, None);
}
#[test]
fn collect_paths_command_refuses_anything_it_did_not_recognize() {
let good = GcCollectTarget {
path: POOL.to_string(),
idle_minutes: 10_080,
trigger: "pooled-ttl",
};
assert!(collect_paths_command(std::slice::from_ref(&good)).is_ok());
assert!(collect_paths_command(&[]).is_err());
for bad_path in [
"/",
"/tmp",
"relative/.rch-target-w-pool-a",
"/tmp/../etc/.rch-target-w-pool-a",
"/tmp/x/.rch-target-w-pool-a; rm -rf /",
"/data/projects/repo/target",
"/data/projects/repo",
] {
let t = GcCollectTarget {
path: bad_path.to_string(),
..good.clone()
};
assert!(
collect_paths_command(&[t]).is_err(),
"{bad_path} must be refused"
);
}
let zero = GcCollectTarget {
idle_minutes: 0,
..good.clone()
};
assert!(collect_paths_command(&[zero]).is_err());
}
#[test]
fn collect_paths_command_rechecks_every_gate_on_the_worker() {
let cmd = collect_paths_command(&[GcCollectTarget {
path: POOL.to_string(),
idle_minutes: 10_080,
trigger: "pooled-ttl",
}])
.expect("valid target");
assert!(cmd.contains("if [ ! -d \"$d\" ]; then printf 'RCH_GC_SKIP %s missing"));
assert!(cmd.contains("find \"$d\" -mmin -\"$__mins\" -print -quit"));
assert!(cmd.contains("if ! __gc_gates_ok \"$d\"; then printf 'RCH_GC_SKIP %s gate"));
assert!(cmd.contains("rm -rf -- \"$d\""));
assert!(cmd.contains("RCH_WORKER_REAP_METRICS"));
assert!(cmd.contains("__gc_gates_ok() {"));
assert!(cmd.contains("[ \"$(__gc_handles \"$1\")\" = free ] || return 1"));
assert!(cmd.contains("[ \"$(__gc_procs \"$1\")\" = free ] || return 1"));
assert!(cmd.contains("__held_ok\" -ne 1 ]; then printf unknown"));
assert!(cmd.contains("__proc_ok\" -ne 1 ]; then printf unknown"));
}
#[test]
fn scan_roots_reuse_the_creating_prelude() {
let prelude = crate::gc_roots::remote_cargo_home_base_prelude();
for cmd in [
enumerate_targets_command("/data/projects"),
worker_sweep_command("/data/projects", 720, Some(10_080), None),
collect_paths_command(&[GcCollectTarget {
path: POOL.to_string(),
idle_minutes: 10_080,
trigger: "pooled-ttl",
}])
.unwrap(),
] {
assert!(
cmd.contains(&prelude),
"the temp base must come from remote_cargo_home_base_prelude, not a copy"
);
}
let cmd = enumerate_targets_command("/data/projects");
assert!(!cmd.contains("__tmpbase=\"${TMPDIR:-}\""));
}
#[test]
fn enumeration_cannot_exit_nonzero_because_a_temp_file_was_never_made() {
let cmd = enumerate_targets_command("/data/projects");
assert!(
cmd.trim_end().ends_with("exit 0"),
"enumerate must end with an explicit status"
);
for var in ["__held", "__proc"] {
assert!(
!cmd.contains(&format!("[ -n \"${var}\" ] && rm -f")),
"${var} cleanup must be an `if`, not an `&&` whose failure becomes the exit status"
);
assert!(cmd.contains(&format!("if [ -n \"${var}\" ]; then rm -f \"${var}\"; fi")));
}
}
#[test]
fn pooled_gate_runs_after_the_cheap_idle_test() {
let cmd = worker_sweep_command("/data/projects", 720, Some(10_080), None);
let gate = cmd
.find("__gc_gates_ok \"$d\"; then printf 'RCH_GC_SKIP pooled-ttl")
.expect("pooled gate present");
let idle = cmd
.find("find \"$d\" -mmin -10080 -print -quit")
.expect("pooled idle test present");
assert!(idle < gate, "the idle test must precede the gate check");
}
#[test]
fn byte_cap_eviction_is_gated_on_liveness_too() {
let cmd = worker_sweep_command("/data/projects", 720, Some(10_080), Some(1024));
assert!(
cmd.contains("if ! __gc_gates_ok \"$d\"; then printf 'RCH_GC_SKIP cap gate"),
"a disk budget must not evict a dir a build is holding open"
);
assert!(cmd.contains("printf 'RCH_GC_SKIP cap active"));
}
#[test]
fn pooled_sweep_pass_is_gated_on_liveness() {
let cmd = worker_sweep_command("/data/projects", 720, Some(10_080), None);
assert!(
cmd.contains("if ! __gc_gates_ok \"$d\"; then printf 'RCH_GC_SKIP pooled-ttl gate"),
"the daemon sweep must apply the same liveness gates to pooled dirs"
);
assert!(!cmd.contains("rch-cargo-cache-*"));
}
#[test]
fn byte_cap_evicts_oldest_first_and_only_past_the_active_floor() {
let p = policy();
let mut entries = vec![
entry(POOL, 5 * DAY, FREE), entry("/data/projects/r/.rch-target-hz2-pool-b", 2 * DAY, FREE),
entry("/data/projects/r/.rch-target-hz2-pool-c", 60, FREE), ];
for e in &mut entries {
e.kb = 10_000;
}
let verdicts: Vec<GcVerdict> = entries
.iter()
.map(|e| evaluate_gc_candidate(e, NOW, &p))
.collect();
assert!(verdicts.iter().all(|v| !v.is_collect()));
assert!(select_cap_evictions(&entries, &verdicts, NOW, &p, 0).is_empty());
assert!(select_cap_evictions(&entries, &verdicts, NOW, &p, 30_000).is_empty());
let evicted = select_cap_evictions(&entries, &verdicts, NOW, &p, 15_000);
assert_eq!(evicted, vec![0, 1]);
let evicted = select_cap_evictions(&entries, &verdicts, NOW, &p, 1);
assert_eq!(
evicted,
vec![0, 1],
"the active-build floor outranks the cap"
);
}
#[test]
fn byte_cap_still_honours_the_liveness_gates() {
let p = policy();
let mut held = entry(POOL, 5 * DAY, (GateEvidence::Held, GateEvidence::Free));
held.kb = 10_000;
let mut unknown = entry(
"/data/projects/r/.rch-target-hz2-pool-b",
5 * DAY,
(GateEvidence::Unknown, GateEvidence::Unknown),
);
unknown.kb = 10_000;
let entries = vec![held, unknown];
let verdicts: Vec<GcVerdict> = entries
.iter()
.map(|e| evaluate_gc_candidate(e, NOW, &p))
.collect();
assert!(
select_cap_evictions(&entries, &verdicts, NOW, &p, 1).is_empty(),
"a held dir and an unprovable dir must both survive the cap"
);
}
#[test]
fn gc_skip_lines_parse_back() {
let out = "RCH_GC_SKIP pooled-ttl gate /a/b/.rch-target-w-pool-x\n\
RCH_GC_SKIP cache-ttl active /a/b/rch-cargo-cache-w\n\
RCH_GC_SKIP bogus\n";
let skips = parse_gc_skips(out);
assert_eq!(skips.len(), 2);
assert_eq!(skips[0].trigger, "pooled-ttl");
assert_eq!(skips[0].reason, "gate");
assert_eq!(skips[1].path, "/a/b/rch-cargo-cache-w");
}
#[cfg(target_os = "linux")]
#[test]
fn open_descriptors_keep_a_stale_pool_alive_end_to_end() {
use std::process::Command;
let tmp = tempfile::tempdir().unwrap();
let base = tmp.path().join("projects");
let tmpbase = tmp.path().join("scratch");
let free_pool = base.join("repo").join(".rch-target-w1-pool-free");
let held_pool = base.join("repo").join(".rch-target-w1-pool-held");
let cache = tmpbase.join("rch-cargo-cache-w1");
for d in [&free_pool, &held_pool, &cache] {
std::fs::create_dir_all(d).unwrap();
std::fs::write(d.join("artifact.o"), b"xxxx").unwrap();
}
for d in [&free_pool, &held_pool, &cache] {
let ok = Command::new("find")
.arg(d)
.args(["-exec", "touch", "-t", "202001010000", "{}", "+"])
.status()
.map(|s| s.success())
.unwrap_or(false);
if !ok {
return; }
}
let _guard = std::fs::File::open(held_pool.join("artifact.o")).unwrap();
let enumerate = enumerate_targets_command(base.to_str().unwrap());
let out = Command::new("sh")
.arg("-c")
.arg(&enumerate)
.env("TMPDIR", &tmpbase)
.output()
.expect("enumerate should execute");
let stdout = String::from_utf8_lossy(&out.stdout);
assert!(
parse_gate_availability(&stdout).both(),
"both gate snapshots must be available on Linux: {stdout}"
);
let entries = parse_target_entries(&stdout);
let find = |needle: &str| {
entries
.iter()
.find(|e| e.path.contains(needle))
.unwrap_or_else(|| panic!("{needle} not enumerated in {entries:?}"))
};
assert_eq!(find("rch-cargo-cache-w1").class(), GcClass::CargoCache);
assert_eq!(find("pool-held").open_handles, GateEvidence::Held);
assert_eq!(find("pool-free").open_handles, GateEvidence::Free);
let targets: Vec<GcCollectTarget> = [&free_pool, &held_pool, &cache]
.iter()
.map(|d| GcCollectTarget {
path: std::fs::canonicalize(d)
.unwrap()
.to_string_lossy()
.into_owned(),
idle_minutes: MIN_POOLED_IDLE_MINUTES,
trigger: "pooled-ttl",
})
.collect();
let cmd = collect_paths_command_with_registry(
&targets,
tmp.path().join("claims-v1").to_str().unwrap(),
)
.expect("targets are valid");
let out = Command::new("sh")
.arg("-c")
.arg(&cmd)
.env("TMPDIR", &tmpbase)
.output()
.expect("collect should execute");
let stdout = String::from_utf8_lossy(&out.stdout);
let stderr = String::from_utf8_lossy(&out.stderr);
assert!(out.status.success(), "collect must be valid sh: {stderr}");
assert!(
!stderr.contains("syntax error") && !stderr.contains("unexpected"),
"collect emitted shell errors: {stderr}"
);
assert!(
!free_pool.exists(),
"an idle, unheld pool must be collected"
);
assert!(
!cache.exists(),
"an idle, unheld cargo cache must be collected"
);
assert!(
held_pool.exists(),
"a pool with an open descriptor must survive: {stdout}"
);
let skipped = parse_gc_skips(&stdout);
assert!(
skipped
.iter()
.any(|s| s.path.contains("pool-held") && s.reason == "gate"),
"the held pool must be reported as gate-skipped: {stdout}"
);
assert_eq!(parse_worker_reap_metrics(&stdout).map(|m| m.0), Some(2));
}
#[test]
fn safe_reap_path_accepts_deep_abs() {
assert!(is_safe_reap_path("/tmp/rch/myproject/abc123"));
assert!(is_safe_reap_path("/tmp/rch"));
}
#[test]
fn safe_reap_path_rejects_dangerous() {
assert!(!is_safe_reap_path("/"));
assert!(!is_safe_reap_path("/tmp")); assert!(!is_safe_reap_path("/tmp/../etc"));
assert!(!is_safe_reap_path("/tmp/rch; rm -rf x"));
assert!(!is_safe_reap_path("relative/path"));
assert!(!is_safe_reap_path("/tmp/$(whoami)"));
}
#[test]
fn safe_reap_base_allows_single_segment_but_not_root() {
assert!(is_safe_reap_base("/srv"));
assert!(is_safe_reap_base("/tmp/rch"));
assert!(!is_safe_reap_base("/"));
assert!(!is_safe_reap_base("//"));
assert!(!is_safe_reap_base("/../x"));
}
#[test]
fn safe_reap_token_rules() {
assert!(is_safe_reap_token(".rch-target-ts2-job-1-2-0"));
assert!(!is_safe_reap_token(""));
assert!(!is_safe_reap_token("."));
assert!(!is_safe_reap_token(".."));
assert!(!is_safe_reap_token("a/b"));
assert!(!is_safe_reap_token("a b"));
}
#[test]
fn idle_minutes_floor() {
assert_eq!(idle_minutes_from_hours(0), 60);
assert_eq!(idle_minutes_from_hours(12), 720);
}
#[test]
fn tmp_base_dot_prefixed_targets_are_discovered() {
let cmd = worker_sweep_command("/data/projects", 720, Some(10_080), Some(1024));
assert!(cmd.contains(&format!(
"find \"$__tmpscan\" -maxdepth {TMPBASE_MAXDEPTH} -type d \\( -name \".rch-target-*-job-*\" -o -name \".rch-target-*-pid-*\" \\)"
)));
assert!(cmd.contains(&format!(
"find \"$__tmpscan\" -maxdepth {TMPBASE_MAXDEPTH} -type d -name \".rch-target-*-pool-*\""
)));
assert!(cmd.contains("-maxdepth 1 -type d -name \"rch_target_*\""));
for var in ["__tmpf", "__tmpf2", "__tmpf3"] {
assert!(
cmd.contains(&format!("sort -u \"${var}\" -o \"$__ded\"")),
"missing dedup for ${var}"
);
assert!(
cmd.contains(&format!("mv -f \"$__ded\" \"${var}\"")),
"dedup for ${var} must move the sorted temp back into place"
);
}
for var in ["__tmpf", "__tmpf2", "__tmpf3"] {
assert!(
!cmd.contains(&format!("sort -u \"${var}\" -o \"${var}\"")),
"in-place sort of ${var} can truncate the candidate list"
);
}
}
#[test]
fn tmp_base_passes_keep_shallow_root_guard() {
let cmd = worker_sweep_command("/data/projects", 720, Some(10_080), Some(1024));
assert!(cmd.contains("case \"$__tmpbase\" in /*/*) __tmpscan="));
assert!(cmd.contains("case \"$__tmpscan\" in /*/*) ;; *) __tmpscan=\"\";; esac"));
assert!(cmd.contains("__rt=$(cd \"$base\" 2>/dev/null && pwd -P)"));
assert!(cmd.contains("__tmpscan=$(cd \"$__tmpbase\" 2>/dev/null && pwd -P)"));
assert!(
!cmd.contains("find \"$__tmpbase\""),
"tmp-base scans must use the resolved $__tmpscan, not $__tmpbase"
);
let scans = cmd.matches("find \"$__tmpscan\"").count();
let gates = cmd.matches("if [ -n \"$__tmpscan\" ]; then").count();
assert!(
scans >= 4,
"expected the job/pid, legacy, pooled and cap scans"
);
assert!(
gates >= 3,
"every tmp-base scan group must be gated on $__tmpscan being set"
);
}
#[test]
fn loop_body_keeps_recent_and_excludes_current() {
let body = reap_loop_body(720, Some(".rch-target-self"), "", "");
assert!(body.contains("[ \"$d\" = \".rch-target-self\" ] && continue"));
assert!(body.contains("find \"$d\" -mmin -720 -print -quit"));
assert!(!body.contains("-type f"));
assert!(body.contains("rm -rf -- \"$d\""));
}
#[test]
fn loop_body_with_metrics_accounts_size() {
let body = reap_loop_body(720, None, "removed", "freed_kb");
assert!(body.contains("du -sk \"$d\""));
assert!(body.contains("removed=$((removed + 1))"));
assert!(body.contains("freed_kb=$((freed_kb + sz))"));
}
fn glob_matches(glob: &str, name: &str) -> bool {
let parts: Vec<&str> = glob.split('*').collect();
let mut pos = 0usize;
for (i, part) in parts.iter().enumerate() {
if part.is_empty() {
continue;
}
match name[pos..].find(part) {
Some(idx) => {
if i == 0 && !glob.starts_with('*') && idx != 0 {
return false;
}
pos += idx + part.len();
}
None => return false,
}
}
if !glob.ends_with('*') {
return name.ends_with(parts.last().copied().unwrap_or(""));
}
true
}
#[test]
fn pooled_target_dir_name_is_reapable() {
let name = ".rch-target-ts2-pool-deadbeefcafef00ddeadbeefcafef00d";
assert!(
REAP_GLOBS.iter().any(|g| glob_matches(g, name)),
"pooled target dir {name} must match a reap glob: {REAP_GLOBS:?}"
);
assert!(
is_safe_reap_token(name),
"pooled name must be reap-token-safe"
);
assert!(
REAP_GLOBS
.iter()
.any(|g| glob_matches(g, ".rch-target-ts2-job-7-123-0"))
);
assert!(
REAP_GLOBS
.iter()
.any(|g| glob_matches(g, ".rch-target-ts2-pid-99-123-0"))
);
assert!(!REAP_GLOBS.iter().any(|g| glob_matches(g, "target")));
assert!(!REAP_GLOBS.iter().any(|g| glob_matches(g, ".rch-target")));
}
#[test]
fn worker_sweep_pooled_pass_is_optional_and_floored() {
let cmd = worker_sweep_command("/data/projects", 720, None, None);
assert!(!cmd.contains("-pool-"));
let cmd = worker_sweep_command("/data/projects", 720, Some(168 * 60), None);
assert!(cmd.contains("-name \".rch-target-*-pool-*\" -prune"));
assert!(
cmd.contains("-mmin -10080"),
"pooled window must reach the predicate"
);
assert!(cmd.contains("done < \"$__tmpf2\""));
assert!(!cmd.contains("| while"));
let cmd = worker_sweep_command("/data/projects", 720, Some(60), None);
assert!(cmd.contains("-mmin -1440"));
}
#[test]
fn worker_sweep_cap_pass_is_optional_and_shaped() {
let cmd = worker_sweep_command("/data/projects", 720, None, None);
assert!(!cmd.contains("total_kb"));
let cmd = worker_sweep_command("/data/projects", 720, None, Some(100 * 1024 * 1024));
assert!(cmd.contains("104857600"));
assert!(cmd.contains("sort -n"));
assert!(cmd.matches("-mmin -720").count() >= 2);
assert!(cmd.contains("done < \"$__tmpf3\""));
assert!(!cmd.contains("| while"));
}
#[test]
fn worker_sweep_command_keeps_the_load_bearing_shape() {
let cmd = worker_sweep_command("/data/projects", 720, None, None);
assert!(cmd.contains(
"find \"$__rt\" -maxdepth 8 -type d \\( -name \".rch-target-*-job-*\" -o -name \".rch-target-*-pid-*\" \\) -prune"
));
assert!(!cmd.contains("sweep-pool"));
assert!(cmd.contains(
"if [ -n \"$__tmpscan\" ]; then find \"$__tmpscan\" -maxdepth 1 -type d -name \"rch_target_*\" -prune"
));
assert!(cmd.contains("done < \"$__tmpf\""));
assert!(!cmd.contains("| while"));
assert!(cmd.contains("RCH_WORKER_REAP_METRICS removed=%s freed_kb=%s"));
assert!(cmd.contains("-mmin -720"));
}
#[test]
fn enumerate_command_is_read_only_and_includes_pooled_dirs() {
let cmd = enumerate_targets_command("/data/projects");
assert!(cmd.contains(
"\\( -name \".rch-target-*-pool-*\" -o -name \"rch-cargo-cache-*\" \\) -prune"
));
assert!(cmd.contains("RCH_TARGET_ENTRY"));
assert!(cmd.contains("done < \"$__tmpf\""));
assert!(!cmd.contains("rm -rf"));
let removals: Vec<&str> = cmd
.match_indices("rm -f ")
.map(|(i, _)| &cmd[i + "rm -f ".len()..])
.collect();
assert!(
!removals.is_empty(),
"enumerate should still clean up its temp files"
);
for tail in removals {
assert!(
tail.starts_with("\"$__"),
"enumerate must only rm its own temp files, found: rm -f {}",
&tail[..tail.len().min(40)]
);
}
assert!(cmd.contains("rm -f \"$__tmpf\""));
}
#[cfg(unix)]
#[test]
fn worker_sweep_executes_and_reaps_only_idle_dirs() {
use std::process::Command;
let tmp = tempfile::tempdir().unwrap();
let base = tmp.path().join("projects");
let tmpbase = tmp.path().join("scratch");
let stale_pool = tmpbase
.join("rch")
.join("proj")
.join("hash")
.join(".rch-target-w1-pool-deadbeef");
let active_pool = tmpbase
.join("rch")
.join("proj2")
.join("hash")
.join(".rch-target-w1-pool-cafe");
let stale_job = base.join("repo").join(".rch-target-w1-job-1-2-0");
let active_job = base.join("repo").join(".rch-target-w1-job-9-9-9");
for d in [&stale_pool, &active_pool, &stale_job, &active_job] {
std::fs::create_dir_all(d).unwrap();
std::fs::write(d.join("artifact.o"), b"xxxx").unwrap();
}
for d in [&stale_pool, &stale_job] {
let ok = Command::new("find")
.arg(d)
.args(["-exec", "touch", "-t", "202001010000", "{}", "+"])
.status()
.map(|s| s.success())
.unwrap_or(false);
if !ok {
return; }
}
let cmd = worker_sweep_command_with_registry(
base.to_str().unwrap(),
60,
Some(MIN_POOLED_IDLE_MINUTES),
None,
tmp.path().join("claims-v1").to_str().unwrap(),
);
let out = Command::new("sh")
.arg("-c")
.arg(&cmd)
.env("TMPDIR", &tmpbase)
.output()
.expect("sweep should execute");
let stderr = String::from_utf8_lossy(&out.stderr);
assert!(
out.status.success(),
"generated sweep must be a valid script; stderr: {stderr}"
);
assert!(
!stderr.contains("syntax error") && !stderr.contains("unexpected"),
"generated sweep emitted shell errors: {stderr}"
);
assert!(
!stale_pool.exists(),
"stale pool under the TMP BASE must be reaped (the regression this fixes)"
);
assert!(
!stale_job.exists(),
"stale job dir under $base must be reaped"
);
assert!(active_pool.exists(), "recently-touched pool must survive");
assert!(active_job.exists(), "recently-touched job dir must survive");
let stdout = String::from_utf8_lossy(&out.stdout);
let (removed, _freed) =
parse_worker_reap_metrics(&stdout).expect("sweep must print its metrics line");
assert_eq!(removed, 2, "exactly the two idle dirs; stdout: {stdout}");
}
#[cfg(target_os = "linux")]
fn source_gc_fixture_dir(path: &std::path::Path) {
std::fs::create_dir_all(path).unwrap();
std::fs::write(path.join("artifact.o"), vec![b'x'; 16 * 1024]).unwrap();
assert!(
std::process::Command::new("find")
.arg(path)
.args(["-exec", "touch", "-t", "202001010000", "{}", "+"])
.status()
.unwrap()
.success()
);
}
#[cfg(target_os = "linux")]
fn source_gc_fixture_claim(
registry: &std::path::Path,
root: &std::path::Path,
token: &str,
extension: &str,
) -> std::path::PathBuf {
std::fs::create_dir_all(registry).unwrap();
let pending = registry.join(format!("input-{token}"));
std::fs::write(&pending, format!("{}\n", root.display())).unwrap();
let output = std::process::Command::new("sha256sum")
.arg(&pending)
.output()
.unwrap();
assert!(output.status.success());
let digest = std::str::from_utf8(&output.stdout)
.unwrap()
.split_whitespace()
.next()
.unwrap();
let record = registry.join(format!("{token}.{digest}.{extension}"));
std::fs::rename(pending, &record).unwrap();
record
}
#[cfg(target_os = "linux")]
#[test]
fn worker_gc_preserves_active_pending_and_aliased_source_claims() {
use std::os::unix::fs::symlink;
use std::process::Command;
let tmp = tempfile::tempdir().unwrap();
let base = tmp.path().join("projects");
let registry = tmp.path().join("claims");
let scratch = tmp.path().join("scratch");
std::fs::create_dir_all(&scratch).unwrap();
let exact = base.join("exact/.rch-target-w-job-exact");
let ancestor = base.join("ancestor/.rch-target-w-job-ancestor");
let descendant = base.join("descendant/.rch-target-w-job-descendant");
let aliased = base.join("aliased/.rch-target-w-job-aliased");
let disjoint = base.join("free/.rch-target-w-job-free");
std::fs::create_dir_all(descendant.join("nested")).unwrap();
for path in [&exact, &ancestor, &descendant, &aliased, &disjoint] {
source_gc_fixture_dir(path);
}
let alias = tmp.path().join("source-alias");
symlink(aliased.parent().unwrap(), &alias).unwrap();
let claims = [
source_gc_fixture_claim(®istry, &exact, "aa", "claim"),
source_gc_fixture_claim(®istry, ancestor.parent().unwrap(), "ab", "claim"),
source_gc_fixture_claim(®istry, &descendant.join("nested"), "ac", "pending"),
source_gc_fixture_claim(®istry, &alias, "ad", "claim"),
];
let command = worker_sweep_command_with_registry(
base.to_str().unwrap(),
60,
None,
None,
registry.to_str().unwrap(),
);
let run = || {
Command::new("sh")
.arg("-c")
.arg(&command)
.env("TMPDIR", &scratch)
.output()
.unwrap()
};
let output = run();
assert!(
output.status.success(),
"{}",
String::from_utf8_lossy(&output.stderr)
);
for path in [&exact, &ancestor, &descendant, &aliased] {
assert!(
path.join("artifact.o").exists(),
"claimed output removed: {path:?}"
);
}
assert!(
!disjoint.exists(),
"a disjoint unowned candidate should be collected"
);
let stdout = String::from_utf8_lossy(&output.stdout);
assert_eq!(parse_worker_reap_metrics(&stdout).unwrap().0, 1, "{stdout}");
assert_eq!(
parse_gc_skips(&stdout)
.iter()
.filter(|skip| skip.reason == "source-ownership")
.count(),
4,
"{stdout}"
);
std::fs::create_dir_all(registry.join("released")).unwrap();
std::fs::create_dir_all(registry.join("cancelled")).unwrap();
for claim in claims {
let destination =
if claim.extension().and_then(|extension| extension.to_str()) == Some("pending") {
registry
.join("cancelled")
.join(claim.with_extension("claim").file_name().unwrap())
} else {
registry.join("released").join(claim.file_name().unwrap())
};
std::fs::rename(&claim, destination).unwrap();
}
let output = run();
assert!(output.status.success());
for path in [&exact, &ancestor, &descendant, &aliased] {
assert!(!path.exists(), "released path remained pinned: {path:?}");
}
assert_eq!(
parse_worker_reap_metrics(&String::from_utf8_lossy(&output.stdout))
.unwrap()
.0,
4
);
}
#[cfg(target_os = "linux")]
#[test]
fn worker_gc_corrupt_source_record_fails_closed() {
let tmp = tempfile::tempdir().unwrap();
let base = tmp.path().join("projects");
let scratch = tmp.path().join("scratch");
let registry = tmp.path().join("claims");
std::fs::create_dir_all(&scratch).unwrap();
let candidate = base.join("repo/.rch-target-w-job-unclaimed");
source_gc_fixture_dir(&candidate);
let record = source_gc_fixture_claim(®istry, &tmp.path().join("other"), "aa", "pending");
std::fs::write(record, b"/truncated").unwrap();
let output = std::process::Command::new("sh")
.arg("-c")
.arg(worker_sweep_command_with_registry(
base.to_str().unwrap(),
60,
None,
None,
registry.to_str().unwrap(),
))
.env("TMPDIR", scratch)
.output()
.unwrap();
assert!(output.status.success());
assert!(candidate.join("artifact.o").exists());
let stdout = String::from_utf8_lossy(&output.stdout);
assert_eq!(parse_worker_reap_metrics(&stdout).unwrap().0, 0);
assert!(
parse_gc_skips(&stdout)
.iter()
.any(|skip| skip.reason == "source-ownership")
);
}
#[cfg(target_os = "linux")]
#[test]
fn byte_cap_and_explicit_gc_cannot_override_source_ownership() {
let tmp = tempfile::tempdir().unwrap();
let base = tmp.path().join("projects");
let scratch = tmp.path().join("scratch");
let registry = tmp.path().join("claims");
std::fs::create_dir_all(&scratch).unwrap();
let owned = base.join("owned/.rch-target-w-pool-owned");
let free = base.join("free/.rch-target-w-pool-free");
source_gc_fixture_dir(&owned);
source_gc_fixture_dir(&free);
source_gc_fixture_claim(®istry, &owned, "aa", "claim");
let run = |command: String| {
std::process::Command::new("sh")
.arg("-c")
.arg(command)
.env("TMPDIR", &scratch)
.output()
.unwrap()
};
let output = run(worker_sweep_command_with_registry(
base.to_str().unwrap(),
60,
None,
Some(1),
registry.to_str().unwrap(),
));
assert!(
output.status.success(),
"{}",
String::from_utf8_lossy(&output.stderr)
);
assert!(owned.join("artifact.o").exists());
assert!(!free.exists());
assert_eq!(
parse_worker_reap_metrics(&String::from_utf8_lossy(&output.stdout))
.unwrap()
.0,
1
);
source_gc_fixture_dir(&free);
let targets = [&owned, &free].map(|path| GcCollectTarget {
path: path.to_string_lossy().into_owned(),
idle_minutes: 60,
trigger: "manual",
});
let output =
run(collect_paths_command_with_registry(&targets, registry.to_str().unwrap()).unwrap());
assert!(
output.status.success(),
"{}",
String::from_utf8_lossy(&output.stderr)
);
assert!(owned.join("artifact.o").exists());
assert!(!free.exists());
assert_eq!(
parse_worker_reap_metrics(&String::from_utf8_lossy(&output.stdout))
.unwrap()
.0,
1
);
}
#[cfg(target_os = "linux")]
#[test]
#[allow(clippy::zombie_processes)]
fn source_gc_preserves_ownership_through_hup_and_killed_supervisor() {
use std::os::unix::fs::PermissionsExt;
use std::process::{Command, Stdio};
use std::time::{Duration, Instant};
for signal in ["-HUP", "-KILL"] {
let tmp = tempfile::tempdir().unwrap();
let candidate = tmp.path().join(".rch-target-w-job-interrupted");
let registry = tmp.path().join("claims");
let ready = tmp.path().join("ready");
let resume = tmp.path().join("resume");
let bin = tmp.path().join("bin");
std::fs::create_dir(&bin).unwrap();
source_gc_fixture_dir(&candidate);
let remover = bin.join("rm");
std::fs::write(&remover, b"#!/bin/sh\nexec 8>&-\n: > \"$RCH_GC_TEST_READY\"\ni=0\nwhile [ ! -e \"$RCH_GC_TEST_RESUME\" ]; do i=$((i+1)); [ \"$i\" -lt 1000 ] || exit 9; sleep 0.01; done\nexec /bin/rm \"$@\"\n").unwrap();
std::fs::set_permissions(&remover, std::fs::Permissions::from_mode(0o755)).unwrap();
let script = format!(
"{} __gc_source_begin \"$1\" || exit 1; rm -rf -- \"$1\"; __gc_source_end;",
source_claim_gate_fragment(registry.to_str().unwrap())
);
let mut child = Command::new("sh")
.args(["-c", &script, "gc-lifetime"])
.arg(&candidate)
.env(
"PATH",
format!("{}:{}", bin.display(), std::env::var("PATH").unwrap()),
)
.env("RCH_GC_TEST_READY", &ready)
.env("RCH_GC_TEST_RESUME", &resume)
.stdin(Stdio::null())
.stdout(Stdio::null())
.stderr(Stdio::piped())
.spawn()
.unwrap();
let deadline = Instant::now() + Duration::from_secs(10);
while !ready.exists() && Instant::now() < deadline {
assert!(
child.try_wait().unwrap().is_none(),
"source guard exited before deletion"
);
std::thread::sleep(Duration::from_millis(10));
}
assert!(ready.exists(), "removal child failed to reach barrier");
assert!(
Command::new("kill")
.args([signal, &child.id().to_string()])
.status()
.unwrap()
.success()
);
let contender = || {
Command::new("flock")
.arg("-n")
.arg(registry.join("metadata.lock"))
.arg("true")
.status()
.unwrap()
};
if signal == "-KILL" {
assert!(!child.wait().unwrap().success());
assert!(
contender().success(),
"the killed supervisor should release its kernel lock"
);
let retry = format!(
"{} __gc_source_begin \"$1\"",
source_claim_gate_fragment(registry.to_str().unwrap())
);
assert!(
!Command::new("sh")
.args(["-c", &retry, "gc-retry"])
.arg(&candidate)
.status()
.unwrap()
.success(),
"surviving deletion lost its durable overlap blocker"
);
} else {
assert!(
!contender().success(),
"admission crossed a still-running deletion"
);
}
assert!(candidate.exists());
let release_claims = || {
let script =
abandoned_gc_claims_command_with_registry(true, registry.to_str().unwrap(), 0);
let output = Command::new("sh").args(["-c", &script]).output().unwrap();
assert!(output.status.success(), "{output:?}");
let (status, claims) =
parse_gc_claim_reports(&String::from_utf8_lossy(&output.stdout));
assert_eq!(status.as_deref(), Some("ok"), "{output:?}");
claims
};
if signal == "-KILL" {
let claims = release_claims();
assert_eq!(claims.len(), 1, "{claims:?}");
assert_eq!(claims[0].verdict, "keep", "{claims:?}");
assert!(
["open-handle", "live-process"].contains(&claims[0].reason.as_str()),
"{claims:?}"
);
}
std::fs::write(resume, b"continue").unwrap();
if signal == "-HUP" {
assert!(child.wait().unwrap().success());
} else {
let deadline = Instant::now() + Duration::from_secs(10);
while candidate.exists() && Instant::now() < deadline {
std::thread::sleep(Duration::from_millis(10));
}
assert!(
std::fs::read_dir(®istry).unwrap().any(|entry| entry
.unwrap()
.path()
.extension()
.is_some_and(|extension| extension == "claim")),
"an interrupted GC claim must remain after its unobserved child completes"
);
let mut claims = release_claims();
let deadline = Instant::now() + Duration::from_secs(10);
while claims.first().is_some_and(|c| c.verdict == "keep")
&& Instant::now() < deadline
{
std::thread::sleep(Duration::from_millis(50));
claims = release_claims();
}
assert_eq!(claims.len(), 1, "{claims:?}");
assert_eq!(claims[0].verdict, "released", "{claims:?}");
assert!(registry.join("released").join(&claims[0].name).is_file());
assert!(release_claims().is_empty());
}
assert!(!candidate.exists());
assert!(
contender().success(),
"metadata lock was not released after deletion"
);
}
}
#[cfg(target_os = "linux")]
#[test]
fn abandoned_gc_claims_respect_age_integrity_and_live_references() {
use std::process::{Command, Stdio};
let tmp = tempfile::tempdir().unwrap();
let registry = tmp.path().join("claims");
let tree = tmp.path().join(".rch-target-w-pool-stranded");
std::fs::create_dir_all(&tree).unwrap();
std::fs::create_dir_all(®istry).unwrap();
let record = format!("{}\n", tree.display());
let staged = tmp.path().join("record");
std::fs::write(&staged, &record).unwrap();
let digest = String::from_utf8(
Command::new("sha256sum")
.arg(&staged)
.output()
.unwrap()
.stdout,
)
.unwrap()
.split_whitespace()
.next()
.unwrap()
.to_string();
let name = format!("fc-00000000-0000-4000-8000-000000000001.{digest}.claim");
let claim = registry.join(&name);
std::fs::write(&claim, &record).unwrap();
let scan = |apply: bool, min_age: u64| {
let script = abandoned_gc_claims_command_with_registry(
apply,
registry.to_str().unwrap(),
min_age,
);
let output = Command::new("sh").args(["-c", &script]).output().unwrap();
assert!(output.status.success(), "{output:?}");
let (status, claims) = parse_gc_claim_reports(&String::from_utf8_lossy(&output.stdout));
assert_eq!(status.as_deref(), Some("ok"), "{output:?}");
claims
};
let only = |claims: Vec<GcClaimReport>| {
assert_eq!(claims.len(), 1, "{claims:?}");
(claims[0].verdict.clone(), claims[0].reason.clone())
};
assert_eq!(only(scan(true, 60)), ("keep".into(), "young".into()));
assert_eq!(
only(scan(false, 0)),
("would-release".into(), "abandoned".into())
);
assert!(claim.is_file(), "a preview must not release");
let mut holder = Command::new("sleep")
.arg("30")
.current_dir(&tree)
.stdin(Stdio::null())
.spawn()
.unwrap();
let held = only(scan(true, 0));
holder.kill().unwrap();
holder.wait().unwrap();
assert_eq!(held, ("keep".into(), "open-handle".into()));
std::fs::write(&claim, b"/tampered\n").unwrap();
assert_eq!(only(scan(true, 0)), ("keep".into(), "unsafe-record".into()));
std::fs::write(&claim, &record).unwrap();
let released = scan(true, 0);
assert_eq!(released[0].path, tree.display().to_string());
assert_eq!(only(released), ("released".into(), "abandoned".into()));
assert!(!claim.exists());
assert!(registry.join("released").join(&name).is_file());
}
#[test]
fn gc_claim_report_lines_parse_with_spaces_in_the_path() {
let (status, claims) = parse_gc_claim_reports(
"noise\nRCH_GC_CLAIMS ok\nRCH_GC_CLAIM keep young 12 fc-a.b.claim /data/my dir\n\
RCH_GC_CLAIM keep x notanumber fc-c.d.claim /p\nRCH_GC_CLAIM short\n",
);
assert_eq!(status.as_deref(), Some("ok"));
assert_eq!(
claims,
vec![GcClaimReport {
verdict: "keep".into(),
reason: "young".into(),
age_minutes: 12,
name: "fc-a.b.claim".into(),
path: "/data/my dir".into(),
}]
);
assert_eq!(parse_gc_claim_reports("").0, None);
}
#[cfg(unix)]
#[test]
fn enumerate_and_sweep_agree_on_tmp_base_discovery() {
use std::process::Command;
let tmp = tempfile::tempdir().unwrap();
let base = tmp.path().join("projects");
let tmpbase = tmp.path().join("scratch");
let tmp_pool = tmpbase
.join("rch")
.join("proj")
.join("hash")
.join(".rch-target-w1-pool-abc");
let base_pool = base.join("repo").join(".rch-target-w1-pool-def");
let tmp_job = tmpbase
.join("rch")
.join("proj")
.join("hash")
.join(".rch-target-w1-job-1-2-0");
for d in [&tmp_pool, &base_pool, &tmp_job] {
std::fs::create_dir_all(d).unwrap();
std::fs::write(d.join("artifact.o"), b"xxxx").unwrap();
}
let cmd = enumerate_targets_command(base.to_str().unwrap());
let out = Command::new("sh")
.arg("-c")
.arg(&cmd)
.env("TMPDIR", &tmpbase)
.output()
.expect("enumerate should execute");
let stderr = String::from_utf8_lossy(&out.stderr);
assert!(out.status.success(), "enumerate must be valid sh: {stderr}");
assert!(
!stderr.contains("syntax error") && !stderr.contains("unexpected"),
"enumerate emitted shell errors: {stderr}"
);
let stdout = String::from_utf8_lossy(&out.stdout);
let entries = parse_target_entries(&stdout);
let listed: Vec<&str> = entries.iter().map(|e| e.path.as_str()).collect();
for expected in [&tmp_pool, &base_pool, &tmp_job] {
let want = std::fs::canonicalize(expected).unwrap_or_else(|_| expected.clone());
let matched = listed.iter().any(|p| {
std::fs::canonicalize(p).map(|c| c == want).unwrap_or(false)
|| std::path::Path::new(p) == want
});
assert!(
matched,
"enumerate must list {}; listed: {listed:?}",
want.display()
);
}
for want in &listed {
assert_eq!(
listed.iter().filter(|p| *p == want).count(),
1,
"duplicate entry for {want} in {listed:?}"
);
}
}
#[cfg(unix)]
#[test]
fn overlapping_roots_are_deduped_not_double_counted() {
use std::process::Command;
let tmp = tempfile::tempdir().unwrap();
let tmpbase = tmp.path().join("scratch");
let base = tmpbase.join("projects");
let stale = base.join("repo").join(".rch-target-w1-job-1-2-0");
std::fs::create_dir_all(&stale).unwrap();
std::fs::write(stale.join("artifact.o"), b"xxxx").unwrap();
let aged = Command::new("find")
.arg(&stale)
.args(["-exec", "touch", "-t", "202001010000", "{}", "+"])
.status()
.map(|s| s.success())
.unwrap_or(false);
if !aged {
return;
}
let cmd = worker_sweep_command_with_registry(
base.to_str().unwrap(),
60,
Some(MIN_POOLED_IDLE_MINUTES),
None,
tmp.path().join("claims-v1").to_str().unwrap(),
);
let out = Command::new("sh")
.arg("-c")
.arg(&cmd)
.env("TMPDIR", &tmpbase)
.output()
.expect("sweep should execute");
assert!(out.status.success());
let stdout = String::from_utf8_lossy(&out.stdout);
let (removed, _) =
parse_worker_reap_metrics(&stdout).expect("sweep must print its metrics line");
assert!(!stale.exists(), "the stale dir should be reaped");
assert_eq!(
removed, 1,
"a dir matched by both passes must be counted once; stdout: {stdout}"
);
}
#[test]
fn parse_worker_reap_metrics_roundtrips() {
let out = "noise\nRCH_WORKER_REAP_METRICS removed=3 freed_kb=204800\n";
assert_eq!(parse_worker_reap_metrics(out), Some((3, 204_800)));
assert!(parse_worker_reap_metrics("no metrics here").is_none());
}
#[test]
fn parse_target_entries_parses_and_skips_garbage() {
let out = "RCH_TARGET_ENTRY 1754700000 1024 /data/tmp/rch_target_old\n\
garbage line\n\
RCH_TARGET_ENTRY notanum 5 /x\n\
RCH_TARGET_ENTRY 1754700001 2048 /data/projects/repo/.rch-target-css-pool-abc\n";
let entries = parse_target_entries(out);
assert_eq!(entries.len(), 2);
assert_eq!(entries[0].path, "/data/tmp/rch_target_old");
assert_eq!(entries[0].kb, 1024);
assert!(!entries[0].is_pooled());
assert!(entries[1].is_pooled());
}
#[test]
fn reap_events_are_emitted_per_pass_and_parse_back() {
let cmd = worker_sweep_command("/data/projects", 720, Some(168 * 60), Some(1024));
assert!(cmd.contains("RCH_REAP_RM %s ttl %s"));
assert!(cmd.contains("RCH_REAP_RM %s pooled-ttl %s"));
assert!(cmd.contains("RCH_REAP_RM %s cap %s"));
let out = "RCH_REAP_RM 2048 ttl /data/projects/r/.rch-target-w-job-1\n\
RCH_REAP_RM 512 cap /data/projects/r/.rch-target-w-pool-x\n\
RCH_WORKER_REAP_METRICS removed=2 freed_kb=2560\n";
let events = parse_reap_events(out);
assert_eq!(events.len(), 2);
assert_eq!(events[0].trigger, "ttl");
assert_eq!(events[1].kb, 512);
assert_eq!(events[1].trigger, "cap");
let hook_body = reap_loop_body(720, None, "", "");
assert!(!hook_body.contains("RCH_REAP_RM"));
}
#[test]
fn rm_failures_are_captured_in_tagged_contexts_only() {
let tagged = reap_loop_body_with_event(720, None, "removed", "freed_kb", Some("ttl"));
assert!(tagged.contains("RCH_REAP_ERR ttl"));
assert!(!tagged.contains("rm -rf -- \"$d\" 2>/dev/null"));
let untagged = reap_loop_body(720, None, "removed", "freed_kb");
assert!(!untagged.contains("RCH_REAP_ERR"));
assert!(untagged.contains("rm -rf -- \"$d\" 2>/dev/null"));
let cmd = worker_sweep_command("/data/projects", 720, None, Some(1024));
assert!(cmd.contains("RCH_REAP_ERR cap"));
assert!(cmd.contains("RCH_REAP_CAP initial_kb=%s cap_kb=1024"));
assert!(cmd.contains("RCH_REAP_CAP final_kb=%s"));
assert!(cmd.contains("RCH_GC_SKIP cap active %s"));
assert!(cmd.contains("RCH_GC_SKIP cap gate %s"));
}
#[test]
fn parse_reap_errors_reads_trigger_path_and_message() {
let out = "RCH_REAP_RM 5 ttl /data/projects/r/.rch-target-w-job-1\n\
RCH_REAP_ERR cap /data/projects/r/.rch-target-w-pool-x :: rm: cannot remove 'x': Permission denied\n\
garbage\n";
let errors = parse_reap_errors(out);
assert_eq!(errors.len(), 1);
assert_eq!(errors[0].trigger, "cap");
assert_eq!(errors[0].path, "/data/projects/r/.rch-target-w-pool-x");
assert!(errors[0].message.contains("Permission denied"));
}
}