From dcfd5112b5d40e2fd206aca421d15512a77b4da3 Mon Sep 17 00:00:00 2001 From: calvin-archastro Date: Wed, 12 Aug 2026 13:38:08 -0700 Subject: [PATCH 1/5] feat: list running service ports --- README.md | 13 +++ src/cli/commands.rs | 16 ++++ src/cli/skills.md | 6 ++ src/dev/mod.rs | 2 + src/dev/plan.rs | 10 +- src/dev/port_allocator.rs | 99 ++++++++++++++++++- src/dev/port_report.rs | 196 ++++++++++++++++++++++++++++++++++++++ src/main.rs | 17 ++++ tests/dev_services.rs | 164 +++++++++++++++++++++++++++++++ 9 files changed, 520 insertions(+), 3 deletions(-) create mode 100644 src/dev/port_report.rs diff --git a/README.md b/README.md index df6fa41..83b6bd3 100644 --- a/README.md +++ b/README.md @@ -327,6 +327,19 @@ aster services logs platform-backend > platform-backend.log Interactive output honors `$PAGER`, then falls back to `less` or `more`. +List every current allocation for this worktree, including separate supervisor +instances and crash-left listeners: + +```console +aster services ports +aster --json services ports +``` + +Human output maps primary ports to service names and includes all dependency +ports. JSON returns a stable `workspace`/`instances` object; each instance has +its supervisor PID, `active` or `orphaned` status, service mappings, and the +complete named-port map. + Clear stale or orphaned processes from development ports before starting the stack again: diff --git a/src/cli/commands.rs b/src/cli/commands.rs index c5ec999..86ccd4f 100644 --- a/src/cli/commands.rs +++ b/src/cli/commands.rs @@ -286,6 +286,9 @@ pub enum ServicesCommands { service: String, }, + /// List ports allocated to running services in this worktree + Ports, + /// Terminate processes listening on configured or specified ports KillPorts { /// Port numbers or configured/allocated names (defaults to all known workspace ports) @@ -396,6 +399,19 @@ mod tests { assert!(Cli::try_parse_from(["aster", "services", "logs", "api", "web"]).is_err()); } + #[test] + fn services_ports_accepts_no_arguments() { + let cli = Cli::try_parse_from(["aster", "services", "ports"]).unwrap(); + assert!(matches!( + cli.command, + Some(Commands::Services { + command: ServicesCommands::Ports + }) + )); + + assert!(Cli::try_parse_from(["aster", "services", "ports", "web"]).is_err()); + } + #[test] fn skills_is_available_without_a_subcommand() { let cli = Cli::try_parse_from(["aster", "--skills"]).unwrap(); diff --git a/src/cli/skills.md b/src/cli/skills.md index 4662735..202fa69 100644 --- a/src/cli/skills.md +++ b/src/cli/skills.md @@ -252,6 +252,8 @@ Interactive `services logs` output honors `$PAGER`, then tries `less` and Configured development ports can be inspected or cleared: ```console +aster services ports +aster --json services ports aster services kill-ports --dry-run aster services kill-ports aster services kill-ports api web 4011 @@ -261,6 +263,10 @@ Named ports come from `[dev.ports]` and this worktree's active or crash-left dynamic allocation manifests. Explicit numeric ports can be inspected or cleared outside an Aster workspace. +`services ports` reports each supervisor instance separately. Human output +maps primary ports to service names; JSON includes `active`/`orphaned` status, +service mappings, and the complete named-port map for scripting. + ## Read target logs and manage the cache Ordinary completed targets have a separate execution-log store: diff --git a/src/dev/mod.rs b/src/dev/mod.rs index c3253d2..70f6508 100644 --- a/src/dev/mod.rs +++ b/src/dev/mod.rs @@ -5,6 +5,7 @@ mod log_files; mod plan; mod port_allocator; mod port_cleanup; +mod port_report; mod process; mod runner; mod tls; @@ -17,5 +18,6 @@ pub use port_cleanup::{ kill_ports, kill_workspace_ports, resolve_port_selection, resolve_workspace_port_selection, KillPortsOptions, }; +pub use port_report::{format_workspace_ports, workspace_ports_report, WorkspacePortsReport}; pub use runner::{run_dev, DevOptions}; pub use tls::{serve_tls, setup_tls}; diff --git a/src/dev/plan.rs b/src/dev/plan.rs index 4d25b5a..330a520 100644 --- a/src/dev/plan.rs +++ b/src/dev/plan.rs @@ -61,7 +61,12 @@ pub fn resolve_dev_plan( .collect(); let selected_control_port = group_control_port.or(config.control_port.as_deref()); let active_ports = collect_active_ports(config, &selected, selected_control_port)?; - let (ports, port_lease) = allocate_dev_ports(workspace_root, config, &active_ports)?; + let selected_services = selected + .iter() + .map(|name| ((*name).to_string(), config.services[*name].port.clone())) + .collect(); + let (ports, port_lease) = + allocate_dev_ports(workspace_root, config, &active_ports, selected_services)?; let control_port = selected_control_port .map(|name| { ports @@ -355,6 +360,7 @@ fn allocate_dev_ports( workspace_root: &Path, config: &DevWorkspaceConfig, active: &HashSet, + services: BTreeMap>, ) -> Result<(HashMap, PortLease)> { let file_env = load_env_files(workspace_root, &config.port_env_files)?; validate_port_offsets(&config.ports)?; @@ -433,7 +439,7 @@ fn allocate_dev_ports( } let ports = resolve_ports(&config.ports, &file_env, &dynamic_values)?; - Ok((ports, allocator.finish(workspace_root)?)) + Ok((ports, allocator.finish(workspace_root, services)?)) } fn dynamic_root(name: &str, configs: &HashMap) -> Result> { diff --git a/src/dev/port_allocator.rs b/src/dev/port_allocator.rs index 145a244..cc260f7 100644 --- a/src/dev/port_allocator.rs +++ b/src/dev/port_allocator.rs @@ -39,6 +39,23 @@ struct AllocationManifest { supervisor_pid: u32, workspace_root: String, ports: BTreeMap, + #[serde(default)] + services: BTreeMap>, +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)] +#[serde(rename_all = "snake_case")] +pub(crate) enum PortAllocationStatus { + Active, + Orphaned, +} + +#[derive(Debug)] +pub(crate) struct WorkspacePortAllocation { + pub supervisor_pid: u32, + pub status: PortAllocationStatus, + pub ports: BTreeMap, + pub services: BTreeMap>, } impl PortAllocator { @@ -129,12 +146,17 @@ impl PortAllocator { Ok(true) } - pub(crate) fn finish(self, workspace_root: &Path) -> Result { + pub(crate) fn finish( + self, + workspace_root: &Path, + services: BTreeMap>, + ) -> Result { let manifest = AllocationManifest { version: 1, supervisor_pid: std::process::id(), workspace_root: canonical_workspace(workspace_root)?, ports: self.named_ports, + services, }; let manifest_path = write_manifest(&self.directory, &manifest)?; Ok(PortLease { @@ -157,6 +179,81 @@ pub(crate) fn workspace_allocated_ports( Ok(ports) } +/// Return live allocations for one worktree. Crash-left manifests whose ports +/// are still occupied are retained as orphaned; fully stale manifests are +/// removed and omitted. +pub(crate) fn workspace_port_allocations( + workspace_root: &Path, +) -> Result> { + let workspace_root = canonical_workspace(workspace_root)?; + let directory = lease_directory(); + if !directory.exists() { + return Ok(Vec::new()); + } + + let allocation_lock = open_lock(&directory.join("allocation.lock"))?; + allocation_lock + .lock_exclusive() + .context("failed to acquire the Aster port allocator lock")?; + + let mut allocations = Vec::new(); + for (path, manifest) in workspace_manifests(&workspace_root)? { + let mut leased = false; + for port in manifest.ports.values().copied().collect::>() { + let file = open_lock(&directory.join(format!("{port}.lock")))?; + match file.try_lock_exclusive() { + Ok(()) => {} + Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => { + leased = true; + break; + } + Err(error) => { + return Err(error) + .with_context(|| format!("failed to inspect port lease for {port}")); + } + } + } + + let occupied = if leased { + false + } else { + manifest + .ports + .values() + .copied() + .map(port_is_available) + .collect::>>()? + .into_iter() + .any(|available| !available) + }; + + let status = if leased { + Some(PortAllocationStatus::Active) + } else if occupied { + Some(PortAllocationStatus::Orphaned) + } else { + fs::remove_file(&path).with_context(|| { + format!( + "failed to remove stale allocation manifest {}", + path.display() + ) + })?; + None + }; + + if let Some(status) = status { + allocations.push(WorkspacePortAllocation { + supervisor_pid: manifest.supervisor_pid, + status, + ports: manifest.ports, + services: manifest.services, + }); + } + } + allocations.sort_by_key(|allocation| allocation.supervisor_pid); + Ok(allocations) +} + /// Remove crash-left manifests only after every recorded lease is unlocked and /// every recorded port is free. Active supervisors retain their manifests. pub(crate) fn prune_workspace_manifests(workspace_root: &Path) -> Result<()> { diff --git a/src/dev/port_report.rs b/src/dev/port_report.rs new file mode 100644 index 0000000..28722f1 --- /dev/null +++ b/src/dev/port_report.rs @@ -0,0 +1,196 @@ +use std::collections::BTreeMap; +use std::fmt::Write; +use std::path::Path; + +use anyhow::{Context, Result}; +use serde::Serialize; + +use crate::config::DevWorkspaceConfig; + +use super::port_allocator::{ + workspace_port_allocations, PortAllocationStatus, WorkspacePortAllocation, +}; + +#[derive(Debug, Serialize)] +pub struct WorkspacePortsReport { + pub workspace: String, + pub instances: Vec, +} + +#[derive(Debug, Serialize)] +pub struct ServicePortInstance { + pub supervisor_pid: u32, + pub status: ServicePortStatus, + pub services: Vec, + pub ports: BTreeMap, +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum ServicePortStatus { + Active, + Orphaned, +} + +#[derive(Debug, Serialize)] +pub struct ServicePort { + pub name: String, + pub port_name: Option, + pub port: Option, +} + +pub fn workspace_ports_report( + workspace_root: &Path, + config: &DevWorkspaceConfig, +) -> Result { + let workspace = workspace_root + .canonicalize() + .with_context(|| { + format!( + "failed to canonicalize workspace root {}", + workspace_root.display() + ) + })? + .to_string_lossy() + .into_owned(); + let instances = workspace_port_allocations(workspace_root)? + .into_iter() + .map(|allocation| report_instance(allocation, config)) + .collect(); + Ok(WorkspacePortsReport { + workspace, + instances, + }) +} + +fn report_instance( + allocation: WorkspacePortAllocation, + config: &DevWorkspaceConfig, +) -> ServicePortInstance { + // Older manifests have no service metadata. Preserve compatibility by + // reconstructing their best available mapping from the current config. + let service_ports = if allocation.services.is_empty() { + config + .services + .iter() + .map(|(name, service)| (name.clone(), service.port.clone())) + .collect::>() + } else { + allocation.services + }; + let mut services = service_ports + .into_iter() + .map(|(name, port_name)| { + let port = port_name + .as_ref() + .and_then(|port_name| allocation.ports.get(port_name)) + .copied(); + ServicePort { + name, + port_name, + port, + } + }) + .collect::>(); + services.sort_by(|left, right| left.name.cmp(&right.name)); + + ServicePortInstance { + supervisor_pid: allocation.supervisor_pid, + status: match allocation.status { + PortAllocationStatus::Active => ServicePortStatus::Active, + PortAllocationStatus::Orphaned => ServicePortStatus::Orphaned, + }, + services, + ports: allocation.ports, + } +} + +pub fn format_workspace_ports(report: &WorkspacePortsReport) -> String { + if report.instances.is_empty() { + return "No running service port allocations found for this worktree.\n".to_string(); + } + + let mut rows = Vec::new(); + for instance in &report.instances { + for (port_name, port) in &instance.ports { + let services = instance + .services + .iter() + .filter(|service| service.port_name.as_deref() == Some(port_name)) + .map(|service| service.name.as_str()) + .collect::>() + .join(","); + rows.push([ + if services.is_empty() { + "-".to_string() + } else { + services + }, + port_name.clone(), + port.to_string(), + match instance.status { + ServicePortStatus::Active => "active".to_string(), + ServicePortStatus::Orphaned => "orphaned".to_string(), + }, + instance.supervisor_pid.to_string(), + ]); + } + for service in instance + .services + .iter() + .filter(|service| service.port_name.is_none()) + { + rows.push([ + service.name.clone(), + "-".to_string(), + "-".to_string(), + match instance.status { + ServicePortStatus::Active => "active".to_string(), + ServicePortStatus::Orphaned => "orphaned".to_string(), + }, + instance.supervisor_pid.to_string(), + ]); + } + } + + let headers = ["SERVICE", "PORT NAME", "PORT", "STATUS", "SUPERVISOR"]; + let widths = std::array::from_fn::<_, 5, _>(|column| { + rows.iter() + .map(|row| row[column].len()) + .chain(std::iter::once(headers[column].len())) + .max() + .unwrap_or(0) + }); + let mut output = String::new(); + writeln!( + output, + "{: Result<()> { return aster::dev::show_service_logs(&workspace_root, &workspace_config.dev, service); } + // Allocation metadata is independent of project discovery and remains + // useful after a supervisor crash leaves service listeners behind. + if let Commands::Services { + command: ServicesCommands::Ports, + } = &command + { + let workspace_config = WorkspaceConfig::load(&workspace_root)?; + let report = aster::dev::workspace_ports_report(&workspace_root, &workspace_config.dev)?; + if output_mode == OutputMode::Json { + output_json(&report)?; + } else if output_mode != OutputMode::Quiet { + print!("{}", aster::dev::format_workspace_ports(&report)); + } + return Ok(()); + } + // TLS setup and serving only need workspace service configuration. Handling // them before discovery lets a supervised TLS target remain independent of // the repository's project graph. @@ -1038,6 +1054,7 @@ fn run() -> Result<()> { )?; } ServicesCommands::Logs { .. } => unreachable!("service logs handled before discovery"), + ServicesCommands::Ports => unreachable!("service ports handled before discovery"), ServicesCommands::Tls { .. } => unreachable!("TLS commands handled before discovery"), }, Commands::RunTarget { ref args } | Commands::ExternalTarget(ref args) => { diff --git a/tests/dev_services.rs b/tests/dev_services.rs index 11c2371..df2a4b0 100644 --- a/tests/dev_services.rs +++ b/tests/dev_services.rs @@ -294,6 +294,49 @@ stream = true assert!(first.try_wait().unwrap().is_none()); assert!(second.try_wait().unwrap().is_none()); + let json_ports = Command::new(env!("CARGO_BIN_EXE_aster")) + .args(["--json", "services", "ports"]) + .current_dir(root) + .env("ASTER_PORT_LEASE_DIR", &lease_dir) + .output() + .unwrap(); + assert!(json_ports.status.success()); + let report: serde_json::Value = serde_json::from_slice(&json_ports.stdout).unwrap(); + assert_eq!( + report["workspace"], + root.canonicalize().unwrap().to_string_lossy().as_ref() + ); + let instances = report["instances"].as_array().unwrap(); + assert_eq!(instances.len(), 2); + for expected_port in [start, start + 1] { + let instance = instances + .iter() + .find(|instance| instance["ports"]["http"] == expected_port) + .unwrap(); + assert_eq!(instance["status"], "active"); + assert_eq!( + instance["ports"]["dependent"], + derived_start + expected_port - start + ); + assert_eq!(instance["services"][0]["name"], "web"); + assert_eq!(instance["services"][0]["port_name"], "http"); + assert_eq!(instance["services"][0]["port"], expected_port); + } + + let human_ports = Command::new(env!("CARGO_BIN_EXE_aster")) + .args(["services", "ports"]) + .current_dir(root) + .env("ASTER_PORT_LEASE_DIR", &lease_dir) + .output() + .unwrap(); + assert!(human_ports.status.success()); + let human_ports = String::from_utf8_lossy(&human_ports.stdout); + assert!(human_ports.contains("SERVICE")); + assert!(human_ports.contains("web")); + assert!(human_ports.contains("dependent")); + assert!(human_ports.contains(&start.to_string())); + assert!(human_ports.contains(&(start + 1).to_string())); + terminate_aster(&mut first); wait_until(Duration::from_secs(5), || { TcpStream::connect(("127.0.0.1", start)).is_err() @@ -314,6 +357,114 @@ stream = true && TcpStream::connect(("127.0.0.1", start + 1)).is_err() && allocation_manifest_count(&lease_dir) == 0 }); + + let empty_ports = Command::new(env!("CARGO_BIN_EXE_aster")) + .args(["services", "ports", "--json"]) + .current_dir(root) + .env("ASTER_PORT_LEASE_DIR", &lease_dir) + .output() + .unwrap(); + assert!(empty_ports.status.success()); + let report: serde_json::Value = serde_json::from_slice(&empty_ports.stdout).unwrap(); + assert!(report["instances"].as_array().unwrap().is_empty()); +} + +#[test] +fn ports_reports_static_and_portless_services_from_the_running_instance() { + let temp = tempfile::tempdir().unwrap(); + let root = temp.path(); + let lease_dir = root.join("leases"); + let reservation = TcpListener::bind(("127.0.0.1", 0)).unwrap(); + let port = reservation.local_addr().unwrap().port(); + drop(reservation); + + fs::create_dir(root.join(".git")).unwrap(); + fs::create_dir(root.join("app")).unwrap(); + fs::write(root.join("app/package.json"), r#"{"name":"app"}"#).unwrap(); + fs::write( + root.join("aster.toml"), + format!( + r#" +[dev.ports.http] +default = {port} + +[dev.services.web] +target = "//app:web" +port = "http" + +[dev.services.worker] +target = "//app:worker" +"# + ), + ) + .unwrap(); + fs::write( + root.join("app/aster.toml"), + r#" +[targets.web] +command = "python3 -m http.server {port}" +stream = true + +[targets.worker] +command = "sleep 30" +stream = true +"#, + ) + .unwrap(); + + let mut supervisor = Command::new(env!("CARGO_BIN_EXE_aster")) + .args(["services", "up", "--no-ui", "--no-watch"]) + .current_dir(root) + .env("ASTER_PORT_LEASE_DIR", &lease_dir) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .unwrap(); + wait_until(Duration::from_secs(20), || { + TcpStream::connect(("127.0.0.1", port)).is_ok() + }); + + let output = Command::new(env!("CARGO_BIN_EXE_aster")) + .args(["services", "ports", "--json"]) + .current_dir(root) + .env("ASTER_PORT_LEASE_DIR", &lease_dir) + .output() + .unwrap(); + assert!(output.status.success()); + let report: serde_json::Value = serde_json::from_slice(&output.stdout).unwrap(); + let instance = &report["instances"][0]; + assert_eq!(instance["status"], "active"); + assert_eq!(instance["ports"]["http"], port); + let services = instance["services"].as_array().unwrap(); + let web = services + .iter() + .find(|service| service["name"] == "web") + .unwrap(); + assert_eq!(web["port_name"], "http"); + assert_eq!(web["port"], port); + let worker = services + .iter() + .find(|service| service["name"] == "worker") + .unwrap(); + assert!(worker["port_name"].is_null()); + assert!(worker["port"].is_null()); + + let human = Command::new(env!("CARGO_BIN_EXE_aster")) + .args(["services", "ports"]) + .current_dir(root) + .env("ASTER_PORT_LEASE_DIR", &lease_dir) + .output() + .unwrap(); + assert!(human.status.success()); + let human = String::from_utf8_lossy(&human.stdout); + assert!(human.lines().any(|line| { + line.contains("web") && line.contains("http") && line.contains(&port.to_string()) + })); + assert!(human + .lines() + .any(|line| line.contains("worker") && line.contains("active"))); + + terminate_aster(&mut supervisor); } #[test] @@ -385,6 +536,19 @@ stream = true }); assert_eq!(allocation_manifest_count(&lease_dir), 1); + let orphaned_ports = Command::new(env!("CARGO_BIN_EXE_aster")) + .args(["--json", "services", "ports"]) + .current_dir(root) + .env("ASTER_PORT_LEASE_DIR", &lease_dir) + .output() + .unwrap(); + assert!(orphaned_ports.status.success()); + let report: serde_json::Value = serde_json::from_slice(&orphaned_ports.stdout).unwrap(); + assert_eq!(report["instances"].as_array().unwrap().len(), 1); + assert_eq!(report["instances"][0]["status"], "orphaned"); + assert_eq!(report["instances"][0]["ports"]["http"], port); + assert_eq!(report["instances"][0]["services"][0]["name"], "web"); + let other_workspace = temp.path().join("other-worktree"); fs::create_dir(&other_workspace).unwrap(); fs::create_dir(other_workspace.join(".git")).unwrap(); From d259e8e82941120049eb6777dd8084dfb2f5a9d9 Mon Sep 17 00:00:00 2001 From: calvin-archastro Date: Wed, 12 Aug 2026 14:17:13 -0700 Subject: [PATCH 2/5] test: serialize service process integration cases --- tests/dev_services.rs | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) diff --git a/tests/dev_services.rs b/tests/dev_services.rs index df2a4b0..2e29ae5 100644 --- a/tests/dev_services.rs +++ b/tests/dev_services.rs @@ -6,6 +6,7 @@ use std::net::{TcpListener, TcpStream}; use std::os::unix::process::CommandExt; use std::path::Path; use std::process::{Command, Stdio}; +use std::sync::{Mutex, MutexGuard, OnceLock}; use std::thread; use std::time::{Duration, Instant}; @@ -27,6 +28,15 @@ fn wait_until(timeout: Duration, condition: impl FnMut() -> bool) { ); } +/// These tests launch real supervisors and listeners. Running them in parallel +/// creates released-port races and can starve process startup on macOS CI. +fn service_process_test() -> MutexGuard<'static, ()> { + static LOCK: OnceLock> = OnceLock::new(); + LOCK.get_or_init(|| Mutex::new(())) + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) +} + fn occurrences(path: &Path, needle: &str) -> usize { fs::read_to_string(path) .unwrap_or_default() @@ -150,6 +160,7 @@ fn terminate_aster(child: &mut std::process::Child) { #[test] fn services_kill_ports_previews_then_clears_configured_listener() { + let _serial = service_process_test(); let temp = tempfile::tempdir().unwrap(); let root = temp.path(); let reservation = TcpListener::bind(("127.0.0.1", 0)).unwrap(); @@ -220,6 +231,7 @@ fn services_kill_ports_previews_then_clears_configured_listener() { #[test] fn dynamic_port_bundles_are_distinct_propagated_and_released() { + let _serial = service_process_test(); let temp = tempfile::tempdir().unwrap(); let root = temp.path(); let lease_dir = root.join("leases"); @@ -371,6 +383,7 @@ stream = true #[test] fn ports_reports_static_and_portless_services_from_the_running_instance() { + let _serial = service_process_test(); let temp = tempfile::tempdir().unwrap(); let root = temp.path(); let lease_dir = root.join("leases"); @@ -469,6 +482,7 @@ stream = true #[test] fn kill_ports_recovers_dynamic_listener_after_supervisor_crash() { + let _serial = service_process_test(); let temp = tempfile::tempdir().unwrap(); let root = temp.path(); let lease_dir = root.join("leases"); @@ -603,6 +617,7 @@ stream = true #[test] fn dev_supervises_targets_runs_prerequisites_and_restarts_on_dependency_changes() { + let _serial = service_process_test(); let temp = tempfile::tempdir().unwrap(); let root = temp.path(); // Keep both reservations open until launch so the OS cannot assign the @@ -795,6 +810,7 @@ command = "sh -c 'echo BUILD >> ../events.log'" #[test] fn dev_restores_through_normal_shutdown_when_every_service_fails_to_start() { + let _serial = service_process_test(); let temp = tempfile::tempdir().unwrap(); let root = temp.path(); let control_port = TcpListener::bind(("127.0.0.1", 0)) @@ -871,6 +887,7 @@ stream = true #[test] fn dev_does_not_start_a_service_after_shutdown_interrupts_its_prerequisite() { + let _serial = service_process_test(); let temp = tempfile::tempdir().unwrap(); let root = temp.path(); fs::create_dir(root.join(".git")).unwrap(); @@ -923,6 +940,7 @@ stream = true #[test] fn authenticated_control_shutdown_interrupts_an_in_progress_prerequisite() { + let _serial = service_process_test(); let temp = tempfile::tempdir().unwrap(); let root = temp.path(); let control_port = TcpListener::bind(("127.0.0.1", 0)) @@ -1075,6 +1093,7 @@ target = "//intern-fe:dev" #[test] fn concurrent_service_groups_bind_distinct_control_ports() { + let _serial = service_process_test(); let temp = tempfile::tempdir().unwrap(); let root = temp.path(); let alpha_reservation = TcpListener::bind(("127.0.0.1", 0)).unwrap(); From c443bbc62a15982aeda5869d61f736c28558a501 Mon Sep 17 00:00:00 2001 From: calvin-archastro Date: Wed, 12 Aug 2026 14:27:59 -0700 Subject: [PATCH 3/5] fix: make port reporting nonblocking --- src/dev/port_allocator.rs | 16 +++++++++------- tests/dev_services.rs | 19 ------------------- 2 files changed, 9 insertions(+), 26 deletions(-) diff --git a/src/dev/port_allocator.rs b/src/dev/port_allocator.rs index cc260f7..0609670 100644 --- a/src/dev/port_allocator.rs +++ b/src/dev/port_allocator.rs @@ -191,11 +191,6 @@ pub(crate) fn workspace_port_allocations( return Ok(Vec::new()); } - let allocation_lock = open_lock(&directory.join("allocation.lock"))?; - allocation_lock - .lock_exclusive() - .context("failed to acquire the Aster port allocator lock")?; - let mut allocations = Vec::new(); for (path, manifest) in workspace_manifests(&workspace_root)? { let mut leased = false; @@ -361,8 +356,15 @@ fn workspace_manifests(workspace_root: &str) -> Result file, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => continue, + Err(error) => { + return Err(error).with_context(|| { + format!("failed to open allocation manifest {}", path.display()) + }); + } + }; let manifest: AllocationManifest = serde_json::from_reader(file) .with_context(|| format!("failed to parse allocation manifest {}", path.display()))?; if manifest.version == 1 && manifest.workspace_root == workspace_root { diff --git a/tests/dev_services.rs b/tests/dev_services.rs index 2e29ae5..df2a4b0 100644 --- a/tests/dev_services.rs +++ b/tests/dev_services.rs @@ -6,7 +6,6 @@ use std::net::{TcpListener, TcpStream}; use std::os::unix::process::CommandExt; use std::path::Path; use std::process::{Command, Stdio}; -use std::sync::{Mutex, MutexGuard, OnceLock}; use std::thread; use std::time::{Duration, Instant}; @@ -28,15 +27,6 @@ fn wait_until(timeout: Duration, condition: impl FnMut() -> bool) { ); } -/// These tests launch real supervisors and listeners. Running them in parallel -/// creates released-port races and can starve process startup on macOS CI. -fn service_process_test() -> MutexGuard<'static, ()> { - static LOCK: OnceLock> = OnceLock::new(); - LOCK.get_or_init(|| Mutex::new(())) - .lock() - .unwrap_or_else(|poisoned| poisoned.into_inner()) -} - fn occurrences(path: &Path, needle: &str) -> usize { fs::read_to_string(path) .unwrap_or_default() @@ -160,7 +150,6 @@ fn terminate_aster(child: &mut std::process::Child) { #[test] fn services_kill_ports_previews_then_clears_configured_listener() { - let _serial = service_process_test(); let temp = tempfile::tempdir().unwrap(); let root = temp.path(); let reservation = TcpListener::bind(("127.0.0.1", 0)).unwrap(); @@ -231,7 +220,6 @@ fn services_kill_ports_previews_then_clears_configured_listener() { #[test] fn dynamic_port_bundles_are_distinct_propagated_and_released() { - let _serial = service_process_test(); let temp = tempfile::tempdir().unwrap(); let root = temp.path(); let lease_dir = root.join("leases"); @@ -383,7 +371,6 @@ stream = true #[test] fn ports_reports_static_and_portless_services_from_the_running_instance() { - let _serial = service_process_test(); let temp = tempfile::tempdir().unwrap(); let root = temp.path(); let lease_dir = root.join("leases"); @@ -482,7 +469,6 @@ stream = true #[test] fn kill_ports_recovers_dynamic_listener_after_supervisor_crash() { - let _serial = service_process_test(); let temp = tempfile::tempdir().unwrap(); let root = temp.path(); let lease_dir = root.join("leases"); @@ -617,7 +603,6 @@ stream = true #[test] fn dev_supervises_targets_runs_prerequisites_and_restarts_on_dependency_changes() { - let _serial = service_process_test(); let temp = tempfile::tempdir().unwrap(); let root = temp.path(); // Keep both reservations open until launch so the OS cannot assign the @@ -810,7 +795,6 @@ command = "sh -c 'echo BUILD >> ../events.log'" #[test] fn dev_restores_through_normal_shutdown_when_every_service_fails_to_start() { - let _serial = service_process_test(); let temp = tempfile::tempdir().unwrap(); let root = temp.path(); let control_port = TcpListener::bind(("127.0.0.1", 0)) @@ -887,7 +871,6 @@ stream = true #[test] fn dev_does_not_start_a_service_after_shutdown_interrupts_its_prerequisite() { - let _serial = service_process_test(); let temp = tempfile::tempdir().unwrap(); let root = temp.path(); fs::create_dir(root.join(".git")).unwrap(); @@ -940,7 +923,6 @@ stream = true #[test] fn authenticated_control_shutdown_interrupts_an_in_progress_prerequisite() { - let _serial = service_process_test(); let temp = tempfile::tempdir().unwrap(); let root = temp.path(); let control_port = TcpListener::bind(("127.0.0.1", 0)) @@ -1093,7 +1075,6 @@ target = "//intern-fe:dev" #[test] fn concurrent_service_groups_bind_distinct_control_ports() { - let _serial = service_process_test(); let temp = tempfile::tempdir().unwrap(); let root = temp.path(); let alpha_reservation = TcpListener::bind(("127.0.0.1", 0)).unwrap(); From bf01cdba44d41c64390139908a06dd9f05c22b39 Mon Sep 17 00:00:00 2001 From: calvin-archastro Date: Wed, 12 Aug 2026 14:35:18 -0700 Subject: [PATCH 4/5] ci: serialize macOS process tests --- .github/workflows/ci.yml | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index f0b18e6..66377b1 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -25,7 +25,13 @@ jobs: steps: - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 - uses: dtolnay/rust-toolchain@4cda84d5c5c54efe2404f9d843567869ab1699d4 - - run: cargo test --locked --all-targets --all-features + # macOS process/listener integration tests contend heavily when libtest + # launches them together; run that runner serially to avoid port races + # and supervisor startup starvation. Linux remains parallel. + - if: runner.os == 'macOS' + run: cargo test --locked --all-targets --all-features -- --test-threads=1 + - if: runner.os != 'macOS' + run: cargo test --locked --all-targets --all-features - run: scripts/test-dynamic-service-ports quality: From ed6914a64952a18d4abd2ac83fc25020bb8b747d Mon Sep 17 00:00:00 2001 From: calvin-archastro Date: Wed, 12 Aug 2026 14:46:27 -0700 Subject: [PATCH 5/5] ci: isolate dynamic port lifecycle test on macOS --- .github/workflows/ci.yml | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 66377b1..b0af22e 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -25,11 +25,15 @@ jobs: steps: - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 - uses: dtolnay/rust-toolchain@4cda84d5c5c54efe2404f9d843567869ab1699d4 - # macOS process/listener integration tests contend heavily when libtest - # launches them together; run that runner serially to avoid port races - # and supervisor startup starvation. Linux remains parallel. + # This multi-supervisor test intentionally releases candidate ports before + # launch. Isolate it on macOS, where concurrent process tests can claim or + # starve those ports long enough to exceed its startup assertion. - if: runner.os == 'macOS' - run: cargo test --locked --all-targets --all-features -- --test-threads=1 + run: | + cargo test --locked --all-targets --all-features -- \ + --skip dynamic_port_bundles_are_distinct_propagated_and_released + cargo test --locked --test dev_services --all-features \ + dynamic_port_bundles_are_distinct_propagated_and_released - if: runner.os != 'macOS' run: cargo test --locked --all-targets --all-features - run: scripts/test-dynamic-service-ports