diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index f0b18e6..b0af22e 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -25,7 +25,17 @@ jobs: steps: - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 - uses: dtolnay/rust-toolchain@4cda84d5c5c54efe2404f9d843567869ab1699d4 - - run: cargo test --locked --all-targets --all-features + # 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 -- \ + --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 quality: 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..0609670 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,76 @@ 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 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<()> { @@ -264,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/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();