| 1 | //! Cross-process admission for expensive local commands. |
| 2 | //! |
| 3 | //! Fleet and Workflow workers execute in separate Codewhale processes, so an |
| 4 | //! in-process semaphore cannot protect the host. Heavy shell commands instead |
| 5 | //! take one of a small number of filesystem-backed permits under |
| 6 | //! `CODEWHALE_HOME`. The default of two permits is deliberately conservative |
| 7 | //! for the 36 GiB laptop class from #4864. |
| 8 | |
| 9 | use std::fs::{File, OpenOptions}; |
| 10 | use std::io; |
| 11 | use std::path::{Path, PathBuf}; |
| 12 | use std::time::{Duration, Instant}; |
| 13 | |
| 14 | use anyhow::{Context, Result, anyhow}; |
| 15 | use fd_lock::{RwLock, RwLockWriteGuard}; |
| 16 | use tokio_util::sync::CancellationToken; |
| 17 | |
| 18 | pub(crate) const DEFAULT_HEAVY_COMMAND_LIMIT: usize = 2; |
| 19 | const MAX_HEAVY_COMMAND_LIMIT: usize = 16; |
| 20 | const ADMISSION_POLL_INTERVAL: Duration = Duration::from_millis(50); |
| 21 | |
| 22 | /// When the host free-RAM fraction drops to/below these thresholds the |
| 23 | /// effective heavy-command admission limit tightens so a saturated host stops |
| 24 | /// admitting new link graphs (#4864 req 7). Values are deliberately generous |
| 25 | /// because the measurement is advisory, not authoritative. |
| 26 | const CONSTRAINED_FREE_FRACTION: f64 = 0.30; |
| 27 | const CRITICAL_FREE_FRACTION: f64 = 0.15; |
| 28 | |
| 29 | #[derive(Debug, Clone, Copy, PartialEq, Eq)] |
| 30 | pub(crate) enum CommandExpense { |
| 31 | Normal, |
| 32 | Heavy, |
| 33 | } |
| 34 | |
| 35 | /// Measured host memory pressure used to tighten heavy-command admission. |
| 36 | /// |
| 37 | /// `Unknown` means "could not be measured"; admission then fails open (uses the |
| 38 | /// configured limit) rather than risk blocking on an unmeasurable host. This |
| 39 | /// keeps the gate safe on any CI runner where the probe is unavailable. |
| 40 | #[derive(Debug, Clone, Copy, PartialEq, Eq)] |
| 41 | pub(crate) enum MemoryPressure { |
| 42 | Unknown, |
| 43 | Nominal, |
| 44 | Constrained, |
| 45 | Critical, |
| 46 | } |
| 47 | |
| 48 | /// Pluggable memory probe so the admission policy is unit-testable without |
| 49 | /// having to drive the host into real memory pressure. |
| 50 | pub(crate) trait MemoryProbe: Send + Sync { |
| 51 | /// Free-RAM fraction in `0.0..=1.0`, or `None` when it cannot be measured. |
| 52 | fn free_fraction(&self) -> Option<f64>; |
| 53 | } |
| 54 | |
| 55 | struct HostMemoryProbe; |
| 56 | |
| 57 | impl MemoryProbe for HostMemoryProbe { |
| 58 | fn free_fraction(&self) -> Option<f64> { |
| 59 | host_memory_free_fraction() |
| 60 | } |
| 61 | } |
| 62 | |
| 63 | fn classify_memory_pressure(free_fraction: Option<f64>) -> MemoryPressure { |
| 64 | match free_fraction { |
| 65 | None => MemoryPressure::Unknown, |
| 66 | Some(fraction) if fraction <= CRITICAL_FREE_FRACTION => MemoryPressure::Critical, |
| 67 | Some(fraction) if fraction <= CONSTRAINED_FREE_FRACTION => MemoryPressure::Constrained, |
| 68 | Some(_) => MemoryPressure::Nominal, |
| 69 | } |
| 70 | } |
| 71 | |
| 72 | /// Effective admission limit after applying host memory pressure. `Critical` |
| 73 | /// yields zero so queued heavy commands wait for the host to recover instead of |
| 74 | /// snowballing; `Constrained` halves the budget (never below one). |
| 75 | fn effective_admission_limit(configured: usize, pressure: MemoryPressure) -> usize { |
| 76 | match pressure { |
| 77 | MemoryPressure::Critical => 0, |
| 78 | MemoryPressure::Constrained => configured.div_ceil(2).max(1), |
| 79 | MemoryPressure::Nominal | MemoryPressure::Unknown => configured, |
| 80 | } |
| 81 | } |
| 82 | |
| 83 | #[derive(Debug)] |
| 84 | struct HeavyPermitSlot { |
| 85 | _guard: RwLockWriteGuard<'static, File>, |
| 86 | // The guard borrows the lock. Keeping the boxed lock here makes that |
| 87 | // allocation outlive the guard; field drop order is guard, then lock. |
| 88 | _lock: Box<RwLock<File>>, |
| 89 | } |
| 90 | |
| 91 | /// A held cross-process heavy-command permit. |
| 92 | #[derive(Debug)] |
| 93 | pub(crate) struct HeavyCommandPermit { |
| 94 | _slot: HeavyPermitSlot, |
| 95 | queued_for: Duration, |
| 96 | limit: usize, |
| 97 | memory_pressure: MemoryPressure, |
| 98 | } |
| 99 | |
| 100 | impl HeavyCommandPermit { |
| 101 | pub(crate) fn queued_for(&self) -> Duration { |
| 102 | self.queued_for |
| 103 | } |
| 104 | |
| 105 | pub(crate) fn limit(&self) -> usize { |
| 106 | self.limit |
| 107 | } |
| 108 | |
| 109 | pub(crate) fn memory_pressure(&self) -> MemoryPressure { |
| 110 | self.memory_pressure |
| 111 | } |
| 112 | } |
| 113 | |
| 114 | pub(crate) fn infer_command_expense(command: &str) -> CommandExpense { |
| 115 | let heavy = command |
| 116 | .split(['\n', '\r', ';', '|', '&']) |
| 117 | .any(segment_is_heavy); |
| 118 | |
| 119 | if heavy { |
| 120 | CommandExpense::Heavy |
| 121 | } else { |
| 122 | CommandExpense::Normal |
| 123 | } |
| 124 | } |
| 125 | |
| 126 | fn segment_is_heavy(segment: &str) -> bool { |
| 127 | let tokens: Vec<String> = segment |
| 128 | .split_whitespace() |
| 129 | .map(|token| token.trim_matches(['"', '\'']).to_string()) |
| 130 | .collect(); |
| 131 | let Some(index) = tokens |
| 132 | .iter() |
| 133 | .position(|token| !token.contains('=') && token != "env") |
| 134 | else { |
| 135 | return false; |
| 136 | }; |
| 137 | let executable = Path::new(&tokens[index]) |
| 138 | .file_stem() |
| 139 | .and_then(|name| name.to_str()) |
| 140 | .unwrap_or_default() |
| 141 | .to_ascii_lowercase(); |
| 142 | if !matches!(executable.as_str(), "cargo" | "rustc") { |
| 143 | return false; |
| 144 | } |
| 145 | if executable == "rustc" { |
| 146 | return true; |
| 147 | } |
| 148 | tokens[index + 1..] |
| 149 | .iter() |
| 150 | .map(|arg| arg.trim().to_ascii_lowercase()) |
| 151 | .find(|arg| !arg.is_empty() && !arg.starts_with('-') && !arg.contains('=')) |
| 152 | .is_some_and(|subcommand| { |
| 153 | matches!( |
| 154 | subcommand.as_str(), |
| 155 | "build" | "test" | "check" | "clippy" | "rustc" |
| 156 | ) |
| 157 | }) |
| 158 | } |
| 159 | |
| 160 | pub(crate) async fn acquire_heavy_command_permit( |
| 161 | command: &str, |
| 162 | cancel: Option<&CancellationToken>, |
| 163 | ) -> Result<Option<HeavyCommandPermit>> { |
| 164 | if infer_command_expense(command) == CommandExpense::Normal { |
| 165 | return Ok(None); |
| 166 | } |
| 167 | |
| 168 | let limit = configured_heavy_command_limit(); |
| 169 | let root = admission_root(); |
| 170 | let probe = HostMemoryProbe; |
| 171 | acquire_heavy_command_permit_at(&root, limit, cancel, &probe) |
| 172 | .await |
| 173 | .map(Some) |
| 174 | } |
| 175 | |
| 176 | async fn acquire_heavy_command_permit_at( |
| 177 | root: &Path, |
| 178 | limit: usize, |
| 179 | cancel: Option<&CancellationToken>, |
| 180 | probe: &dyn MemoryProbe, |
| 181 | ) -> Result<HeavyCommandPermit> { |
| 182 | std::fs::create_dir_all(root) |
| 183 | .with_context(|| format!("creating resource admission directory {}", root.display()))?; |
| 184 | let started = Instant::now(); |
| 185 | |
| 186 | loop { |
| 187 | if cancel.is_some_and(|token| token.is_cancelled()) { |
| 188 | return Err(anyhow!( |
| 189 | "heavy command canceled while queued for resource admission" |
| 190 | )); |
| 191 | } |
| 192 | // Re-measure each iteration: under memory pressure the effective limit |
| 193 | // tightens so a saturated host stops admitting new heavy link graphs |
| 194 | // (#4864 req 7). Critical pressure yields zero slots, so the command |
| 195 | // waits for recovery instead of snowballing. |
| 196 | let pressure = classify_memory_pressure(probe.free_fraction()); |
| 197 | let effective = effective_admission_limit(limit, pressure); |
| 198 | for slot in 0..effective { |
| 199 | let path = root.join(format!("heavy-{slot}.lock")); |
| 200 | match try_lock_slot(&path) { |
| 201 | Ok(Some(slot)) => { |
| 202 | return Ok(HeavyCommandPermit { |
| 203 | _slot: slot, |
| 204 | queued_for: started.elapsed(), |
| 205 | limit, |
| 206 | memory_pressure: pressure, |
| 207 | }); |
| 208 | } |
| 209 | Ok(None) => {} |
| 210 | Err(error) => { |
| 211 | return Err(error).with_context(|| { |
| 212 | format!("acquiring heavy command permit {}", path.display()) |
| 213 | }); |
| 214 | } |
| 215 | } |
| 216 | } |
| 217 | tokio::time::sleep(ADMISSION_POLL_INTERVAL).await; |
| 218 | } |
| 219 | } |
| 220 | |
| 221 | fn try_lock_slot(path: &Path) -> io::Result<Option<HeavyPermitSlot>> { |
| 222 | let file = OpenOptions::new() |
| 223 | .create(true) |
| 224 | .read(true) |
| 225 | .write(true) |
| 226 | .truncate(false) |
| 227 | .open(path)?; |
| 228 | let lock = Box::new(RwLock::new(file)); |
| 229 | let lock_ptr = Box::into_raw(lock); |
| 230 | // SAFETY: `lock_ptr` remains allocated in `HeavyPermitSlot::_lock` for the |
| 231 | // lifetime of `_guard`, and the guard is dropped before that box. |
| 232 | let guard = match unsafe { (&mut *lock_ptr).try_write() } { |
| 233 | Ok(guard) => guard, |
| 234 | Err(error) if error.kind() == io::ErrorKind::WouldBlock => { |
| 235 | // SAFETY: no guard was created, so reclaim the allocation now. |
| 236 | unsafe { drop(Box::from_raw(lock_ptr)) }; |
| 237 | return Ok(None); |
| 238 | } |
| 239 | Err(error) => { |
| 240 | // SAFETY: no guard was created, so reclaim the allocation now. |
| 241 | unsafe { drop(Box::from_raw(lock_ptr)) }; |
| 242 | return Err(error); |
| 243 | } |
| 244 | }; |
| 245 | // SAFETY: the allocation is owned exactly once by this box and is stable on |
| 246 | // the heap even if `HeavyPermitSlot` moves. |
| 247 | let lock = unsafe { Box::from_raw(lock_ptr) }; |
| 248 | // SAFETY: the boxed lock remains alive until after `_guard` is dropped. |
| 249 | let guard = unsafe { |
| 250 | std::mem::transmute::<RwLockWriteGuard<'_, File>, RwLockWriteGuard<'static, File>>(guard) |
| 251 | }; |
| 252 | Ok(Some(HeavyPermitSlot { |
| 253 | _guard: guard, |
| 254 | _lock: lock, |
| 255 | })) |
| 256 | } |
| 257 | |
| 258 | fn configured_heavy_command_limit() -> usize { |
| 259 | std::env::var("CODEWHALE_HEAVY_COMMAND_LIMIT") |
| 260 | .ok() |
| 261 | .and_then(|value| value.trim().parse::<usize>().ok()) |
| 262 | .filter(|limit| *limit > 0) |
| 263 | .unwrap_or(DEFAULT_HEAVY_COMMAND_LIMIT) |
| 264 | .min(MAX_HEAVY_COMMAND_LIMIT) |
| 265 | } |
| 266 | |
| 267 | fn admission_root() -> PathBuf { |
| 268 | if let Some(home) = codewhale_paths::codewhale_home_override().ok().flatten() { |
| 269 | return home.join("resource-admission"); |
| 270 | } |
| 271 | if let Some(home) = codewhale_paths::user_home() { |
| 272 | return home.join(".codewhale").join("resource-admission"); |
| 273 | } |
| 274 | std::env::temp_dir().join("codewhale-resource-admission") |
| 275 | } |
| 276 | |
| 277 | /// Host free-RAM fraction in `0.0..=1.0`, or `None` when it cannot be measured. |
| 278 | /// |
| 279 | /// Each implementation shells out to a standard, always-present tool so no new |
| 280 | /// crate or build-feature dependency is introduced, and every error path returns |
| 281 | /// `None` so admission fails open (never blocks on an unmeasurable host). This |
| 282 | /// keeps the gate safe on any CI runner, while protecting the macOS dogfood host |
| 283 | /// and Linux/Windows machines where the tool exists. |
| 284 | #[cfg(target_os = "linux")] |
| 285 | fn host_memory_free_fraction() -> Option<f64> { |
| 286 | let meminfo = std::fs::read_to_string("/proc/meminfo").ok()?; |
| 287 | let total = parse_meminfo_kb(&meminfo, "MemTotal:")?; |
| 288 | let available = parse_meminfo_kb(&meminfo, "MemAvailable:")?; |
| 289 | (total > 0).then(|| (available as f64 / total as f64).clamp(0.0, 1.0)) |
| 290 | } |
| 291 | |
| 292 | #[cfg(target_os = "linux")] |
| 293 | fn parse_meminfo_kb(meminfo: &str, prefix: &str) -> Option<u64> { |
| 294 | meminfo |
| 295 | .lines() |
| 296 | .find(|line| line.starts_with(prefix)) |
| 297 | .and_then(|line| line.split_whitespace().nth(1)) |
| 298 | .and_then(|value| value.parse::<u64>().ok()) |
| 299 | } |
| 300 | |
| 301 | #[cfg(target_os = "macos")] |
| 302 | fn host_memory_free_fraction() -> Option<f64> { |
| 303 | let total = run_capture("/usr/sbin/sysctl", &["-n", "hw.memsize"]) |
| 304 | .and_then(|bytes| bytes.trim().parse::<u64>().ok())?; |
| 305 | let page_size = run_capture("/usr/bin/pagesize", &[])? |
| 306 | .trim() |
| 307 | .parse::<u64>() |
| 308 | .ok()?; |
| 309 | let stats = run_capture("/usr/bin/vm_stat", &[])?; |
| 310 | let free_pages = memory_pages_from_vm_stat(&stats, &["Pages free:", "Pages inactive:"])?; |
| 311 | let free_bytes = free_pages.checked_mul(page_size)?; |
| 312 | (total > 0).then(|| (free_bytes as f64 / total as f64).clamp(0.0, 1.0)) |
| 313 | } |
| 314 | |
| 315 | #[cfg(target_os = "macos")] |
| 316 | fn memory_pages_from_vm_stat(vm_stat: &str, prefixes: &[&str]) -> Option<u64> { |
| 317 | let mut total = 0u64; |
| 318 | for prefix in prefixes { |
| 319 | let pages = vm_stat |
| 320 | .lines() |
| 321 | .find(|line| line.trim_start().starts_with(prefix)) |
| 322 | .and_then(|line| { |
| 323 | line.split('.') |
| 324 | .nth(1) |
| 325 | .and_then(|rest| rest.trim().parse::<u64>().ok()) |
| 326 | })?; |
| 327 | total = total.checked_add(pages)?; |
| 328 | } |
| 329 | Some(total) |
| 330 | } |
| 331 | |
| 332 | #[cfg(windows)] |
| 333 | fn host_memory_free_fraction() -> Option<f64> { |
| 334 | // `wmic` is deprecated but present on every supported Windows runner and |
| 335 | // avoids adding a GlobalMemoryStatusEx build dependency. Fail open on error. |
| 336 | let out = run_capture( |
| 337 | "C:\\Windows\\System32\\wbem\\wmic.exe", |
| 338 | &[ |
| 339 | "OS", |
| 340 | "get", |
| 341 | "FreePhysicalMemory,TotalVisibleMemorySize", |
| 342 | "/value", |
| 343 | ], |
| 344 | )?; |
| 345 | let free_kb = wmic_value(&out, "FreePhysicalMemory=")?; |
| 346 | let total_kb = wmic_value(&out, "TotalVisibleMemorySize=")?; |
| 347 | (total_kb > 0).then(|| (free_kb as f64 / total_kb as f64).clamp(0.0, 1.0)) |
| 348 | } |
| 349 | |
| 350 | #[cfg(windows)] |
| 351 | fn wmic_value(output: &str, key: &str) -> Option<u64> { |
| 352 | output |
| 353 | .lines() |
| 354 | .find_map(|line| line.trim().strip_prefix(key)) |
| 355 | .and_then(|value| value.trim().parse::<u64>().ok()) |
| 356 | } |
| 357 | |
| 358 | #[cfg(not(any(target_os = "linux", target_os = "macos", target_os = "windows")))] |
| 359 | fn host_memory_free_fraction() -> Option<f64> { |
| 360 | None |
| 361 | } |
| 362 | |
| 363 | #[cfg(any(target_os = "macos", target_os = "windows"))] |
| 364 | fn run_capture(program: &str, args: &[&str]) -> Option<String> { |
| 365 | let output = std::process::Command::new(program) |
| 366 | .args(args) |
| 367 | .output() |
| 368 | .ok()?; |
| 369 | if !output.status.success() { |
| 370 | return None; |
| 371 | } |
| 372 | String::from_utf8(output.stdout).ok() |
| 373 | } |
| 374 | |
| 375 | #[cfg(test)] |
| 376 | mod tests { |
| 377 | use std::sync::Arc; |
| 378 | use std::sync::atomic::{AtomicUsize, Ordering}; |
| 379 | |
| 380 | use super::*; |
| 381 | |
| 382 | /// Deterministic probe returning a fixed fraction so admission behavior is |
| 383 | /// independent of the host running the test suite. |
| 384 | struct StaticMemoryProbe(Option<f64>); |
| 385 | impl MemoryProbe for StaticMemoryProbe { |
| 386 | fn free_fraction(&self) -> Option<f64> { |
| 387 | self.0 |
| 388 | } |
| 389 | } |
| 390 | |
| 391 | const NOMINAL_PROBE: StaticMemoryProbe = StaticMemoryProbe(Some(0.9)); |
| 392 | |
| 393 | #[test] |
| 394 | fn whitespace_codewhale_home_uses_shared_user_home_for_admission_state() { |
| 395 | let _lock = crate::test_support::lock_test_env(); |
| 396 | let tmp = tempfile::tempdir().expect("temporary root"); |
| 397 | let home = tmp.path().join("home"); |
| 398 | let userprofile = tmp.path().join("userprofile"); |
| 399 | let _home = crate::test_support::EnvVarGuard::set("HOME", &home); |
| 400 | let _userprofile = crate::test_support::EnvVarGuard::set("USERPROFILE", &userprofile); |
| 401 | let _codewhale_home = crate::test_support::EnvVarGuard::set("CODEWHALE_HOME", " \t "); |
| 402 | |
| 403 | assert_eq!( |
| 404 | admission_root(), |
| 405 | home.join(".codewhale").join("resource-admission") |
| 406 | ); |
| 407 | } |
| 408 | |
| 409 | #[test] |
| 410 | fn memory_pressure_classification_and_effective_limit() { |
| 411 | // Unknown (unmeasurable) fails open to the configured limit. |
| 412 | assert_eq!(classify_memory_pressure(None), MemoryPressure::Unknown); |
| 413 | assert_eq!(effective_admission_limit(2, MemoryPressure::Unknown), 2); |
| 414 | assert_eq!(classify_memory_pressure(Some(0.9)), MemoryPressure::Nominal); |
| 415 | assert_eq!( |
| 416 | classify_memory_pressure(Some(0.30)), |
| 417 | MemoryPressure::Constrained |
| 418 | ); |
| 419 | assert_eq!( |
| 420 | classify_memory_pressure(Some(0.15)), |
| 421 | MemoryPressure::Critical |
| 422 | ); |
| 423 | assert_eq!( |
| 424 | classify_memory_pressure(Some(0.0)), |
| 425 | MemoryPressure::Critical |
| 426 | ); |
| 427 | assert_eq!(effective_admission_limit(4, MemoryPressure::Nominal), 4); |
| 428 | assert_eq!(effective_admission_limit(4, MemoryPressure::Constrained), 2); |
| 429 | assert_eq!(effective_admission_limit(1, MemoryPressure::Constrained), 1); |
| 430 | assert_eq!(effective_admission_limit(4, MemoryPressure::Critical), 0); |
| 431 | } |
| 432 | |
| 433 | #[test] |
| 434 | fn infers_only_expensive_rust_compilation_commands() { |
| 435 | for command in [ |
| 436 | "cargo test -p codewhale-tui shell::tests", |
| 437 | "env CARGO_BUILD_JOBS=2 cargo build --workspace", |
| 438 | "cargo check", |
| 439 | "cargo clippy --all-targets", |
| 440 | "/usr/bin/rustc src/main.rs", |
| 441 | "printf ok && cargo rustc -- --emit=metadata", |
| 442 | ] { |
| 443 | assert_eq!( |
| 444 | infer_command_expense(command), |
| 445 | CommandExpense::Heavy, |
| 446 | "{command}" |
| 447 | ); |
| 448 | } |
| 449 | for command in [ |
| 450 | "cargo fmt --check", |
| 451 | "cargo metadata", |
| 452 | "git status", |
| 453 | "echo cargo test", |
| 454 | ] { |
| 455 | assert_eq!( |
| 456 | infer_command_expense(command), |
| 457 | CommandExpense::Normal, |
| 458 | "{command}" |
| 459 | ); |
| 460 | } |
| 461 | } |
| 462 | |
| 463 | #[tokio::test(flavor = "multi_thread", worker_threads = 4)] |
| 464 | async fn cross_task_heavy_admission_never_exceeds_limit() { |
| 465 | let temp = tempfile::tempdir().expect("tempdir"); |
| 466 | let active = Arc::new(AtomicUsize::new(0)); |
| 467 | let peak = Arc::new(AtomicUsize::new(0)); |
| 468 | let mut tasks = Vec::new(); |
| 469 | |
| 470 | for _ in 0..12 { |
| 471 | let root = temp.path().to_path_buf(); |
| 472 | let active = Arc::clone(&active); |
| 473 | let peak = Arc::clone(&peak); |
| 474 | tasks.push(tokio::spawn(async move { |
| 475 | let _permit = acquire_heavy_command_permit_at(&root, 2, None, &NOMINAL_PROBE) |
| 476 | .await |
| 477 | .expect("permit"); |
| 478 | let current = active.fetch_add(1, Ordering::SeqCst) + 1; |
| 479 | peak.fetch_max(current, Ordering::SeqCst); |
| 480 | tokio::time::sleep(Duration::from_millis(30)).await; |
| 481 | active.fetch_sub(1, Ordering::SeqCst); |
| 482 | })); |
| 483 | } |
| 484 | for task in tasks { |
| 485 | task.await.expect("admission task"); |
| 486 | } |
| 487 | |
| 488 | assert_eq!(active.load(Ordering::SeqCst), 0); |
| 489 | assert_eq!(peak.load(Ordering::SeqCst), 2); |
| 490 | } |
| 491 | |
| 492 | #[tokio::test] |
| 493 | async fn queued_admission_observes_cancellation() { |
| 494 | let temp = tempfile::tempdir().expect("tempdir"); |
| 495 | let _held = acquire_heavy_command_permit_at(temp.path(), 1, None, &NOMINAL_PROBE) |
| 496 | .await |
| 497 | .expect("initial permit"); |
| 498 | let cancel = CancellationToken::new(); |
| 499 | let wait_cancel = cancel.clone(); |
| 500 | let root = temp.path().to_path_buf(); |
| 501 | let waiter = tokio::spawn(async move { |
| 502 | acquire_heavy_command_permit_at(&root, 1, Some(&wait_cancel), &NOMINAL_PROBE).await |
| 503 | }); |
| 504 | |
| 505 | tokio::time::sleep(Duration::from_millis(75)).await; |
| 506 | cancel.cancel(); |
| 507 | let error = tokio::time::timeout(Duration::from_secs(1), waiter) |
| 508 | .await |
| 509 | .expect("bounded cancellation") |
| 510 | .expect("waiter task") |
| 511 | .expect_err("queued command must cancel"); |
| 512 | assert!(error.to_string().contains("canceled while queued")); |
| 513 | } |
| 514 | |
| 515 | #[tokio::test] |
| 516 | async fn critical_memory_pressure_queues_without_admitting() { |
| 517 | let temp = tempfile::tempdir().expect("tempdir"); |
| 518 | // Critical pressure -> zero effective slots -> the command cannot be |
| 519 | // admitted and must observe cancellation rather than spin forever. |
| 520 | let critical = StaticMemoryProbe(Some(0.05)); |
| 521 | let cancel = CancellationToken::new(); |
| 522 | let wait_cancel = cancel.clone(); |
| 523 | let root = temp.path().to_path_buf(); |
| 524 | let waiter = tokio::spawn(async move { |
| 525 | acquire_heavy_command_permit_at(&root, 2, Some(&wait_cancel), &critical).await |
| 526 | }); |
| 527 | tokio::time::sleep(Duration::from_millis(120)).await; |
| 528 | cancel.cancel(); |
| 529 | let error = tokio::time::timeout(Duration::from_secs(2), waiter) |
| 530 | .await |
| 531 | .expect("bounded cancellation") |
| 532 | .expect("waiter task") |
| 533 | .expect_err("critical-pressure command must not be admitted"); |
| 534 | assert!(error.to_string().contains("canceled while queued")); |
| 535 | } |
| 536 | |
| 537 | #[test] |
| 538 | fn host_memory_probe_is_fail_safe() { |
| 539 | // On any supported host the probe either measures a plausible fraction |
| 540 | // or admits it cannot; it must never panic or return an out-of-range. |
| 541 | if let Some(fraction) = host_memory_free_fraction() { |
| 542 | assert!( |
| 543 | (0.0..=1.0).contains(&fraction), |
| 544 | "measured free fraction out of range: {fraction}" |
| 545 | ); |
| 546 | } |
| 547 | } |
| 548 | |
| 549 | /// Opt-in real cargo acceptance (#4864 req 8): drive an actual heavy-class |
| 550 | /// cargo invocation through the admission path end to end. `cargo check |
| 551 | /// --version` classifies as heavy (subcommand `check`) but exits instantly, |
| 552 | /// so this is fast. Gated behind an env var so CI never runs a real cargo. |
| 553 | #[cfg(unix)] |
| 554 | #[tokio::test] |
| 555 | async fn real_cargo_command_is_admitted_and_released() { |
| 556 | if std::env::var_os("CODEWHALE_RESOURCE_ADMISSION_RUST_ACCEPTANCE").is_none() { |
| 557 | tracing::info!( |
| 558 | "skipping opt-in real-rust admission acceptance; \ |
| 559 | set CODEWHALE_RESOURCE_ADMISSION_RUST_ACCEPTANCE=1 to run" |
| 560 | ); |
| 561 | return; |
| 562 | } |
| 563 | let temp = tempfile::tempdir().expect("tempdir"); |
| 564 | let permit = acquire_heavy_command_permit_at(temp.path(), 2, None, &NOMINAL_PROBE) |
| 565 | .await |
| 566 | .expect("heavy permit for real cargo"); |
| 567 | assert_eq!(permit.limit(), 2); |
| 568 | assert_eq!(permit.memory_pressure(), MemoryPressure::Nominal); |
| 569 | let cargo = std::process::Command::new("cargo") |
| 570 | .arg("check") |
| 571 | .arg("--version") |
| 572 | .output() |
| 573 | .expect("run cargo"); |
| 574 | assert!(cargo.status.success(), "cargo check --version failed"); |
| 575 | drop(permit); |
| 576 | let _again = acquire_heavy_command_permit_at(temp.path(), 2, None, &NOMINAL_PROBE) |
| 577 | .await |
| 578 | .expect("re-acquire after release"); |
| 579 | } |
| 580 | } |
| 581 |