From a346ba098472da61ff68687f30e6ac82fe9ea76e Mon Sep 17 00:00:00 2001 From: Daniel Vianna <1708810+pasunboneleve@users.noreply.github.com> Date: Tue, 1 Sep 2026 16:50:03 +1000 Subject: [PATCH 1/3] Fail once when a managed startup port is occupied Context: Two devloop sessions could target the same managed server port. The second session repeatedly restarted its failed child and kept the watcher alive, leaving an unhealthy supervisor on the occupied address. Decision: Treat a loopback HTTP readiness endpoint as the named process's owned startup address. Check bind availability before launch, classify a post-launch bind race with the same typed error, suppress startup restart effects, clean up the attempted session, and exit non-zero with one filter-independent diagnostic. Preserve concurrent sessions on OS-assigned or otherwise distinct ports. Alternatives considered: A repository singleton lock would prevent legitimate concurrent worktrees. Adding a second bind-address field would duplicate the existing process readiness endpoint and permit contradictory configuration. Tradeoffs: Only loopback HTTP readiness probes provide an address devloop can own and diagnose generically. State-key and remote readiness probes retain their existing process-exit and timeout errors. Architectural impact: Runtime workflow effects now carry startup versus runtime origin. That origin controls whether a failed first-readiness attempt may apply the configured restart policy, while later recoverable workflows keep their existing restart behavior. Runtime failures are written and printed independently of tracing filters. --- CHANGELOG.md | 7 ++ docs/behavior.md | 13 +- fixtures/port-session/server.py | 9 ++ src/core.rs | 20 ++- src/engine.rs | 95 ++++++++++++-- src/main.rs | 29 ++++- src/processes.rs | 148 +++++++++++++++++++--- tests/port_collisions.rs | 215 ++++++++++++++++++++++++++++++++ 8 files changed, 503 insertions(+), 33 deletions(-) create mode 100644 fixtures/port-session/server.py create mode 100644 tests/port_collisions.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 1d856ea..16cffd8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,13 @@ All notable changes to `devloop` will be recorded in this file. ## [Unreleased] +### Fixed + +- Made startup stop with a clear non-zero error when a managed process's local + readiness address is already occupied, without restarting the failed process + or disturbing the existing listener. Concurrent sessions remain supported + when they use different ports. + ## [0.10.3] - 2026-08-26 ### Fixed diff --git a/docs/behavior.md b/docs/behavior.md index c767286..2c01e98 100644 --- a/docs/behavior.md +++ b/docs/behavior.md @@ -43,6 +43,14 @@ When `devloop run` starts, it: 7. runs each workflow named in `startup_workflows` in order 8. starts watching the configured `root` +Before starting a managed process with a loopback HTTP readiness probe, +`devloop` checks whether it can bind that process-owned readiness address. An +occupied address, including one that is bound but not yet accepting +connections, stops startup with a non-zero error that names the process and +address. The colliding session does not enter its restart loop or disturb the +existing listener. This check is per configured address: concurrent sessions +and worktrees remain independent when they use different ports. + The in-memory session state is authoritative for the running process. Edits made directly to the JSON file while `devloop` is running are not merged back into the live session. @@ -124,7 +132,10 @@ Managed processes are long-running child commands. process stops or exits, so a failed dependency cannot leave a stale URL or other process-derived value available to later workflows. - `restart = "always"` restarts a child after any exit unless - `devloop` is shutting down. + `devloop` is shutting down. A child that exits while a startup workflow is + waiting for its first readiness result is treated as a failed start and is + not restarted before that startup failure terminates the session. Later + runtime workflows preserve the configured restart policy. - `restart = "on_failure"` restarts only after unsuccessful exit. - `restart = "never"` never restarts automatically. - Restart policies use the managed command's exit status. A wrapper that diff --git a/fixtures/port-session/server.py b/fixtures/port-session/server.py new file mode 100644 index 0000000..e762c22 --- /dev/null +++ b/fixtures/port-session/server.py @@ -0,0 +1,9 @@ +import http.server +import sys + + +host = "127.0.0.1" +port = int(sys.argv[1]) +server = http.server.ThreadingHTTPServer((host, port), http.server.SimpleHTTPRequestHandler) +print(f"listening {server.server_address[1]}", flush=True) +server.serve_forever() diff --git a/src/core.rs b/src/core.rs index 24340fc..b6098b7 100644 --- a/src/core.rs +++ b/src/core.rs @@ -100,6 +100,7 @@ pub enum RuntimeEffect { RunWorkflow { workflow_name: String, changed_files: Vec, + origin: WorkflowRunOrigin, }, StartWatching, MaintainProcesses, @@ -115,6 +116,12 @@ pub enum RuntimeEffect { Exit, } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum WorkflowRunOrigin { + Startup, + Runtime, +} + #[derive(Debug, Clone)] struct ObservedHookRuntimeState { workflow_name: String, @@ -371,6 +378,7 @@ impl RuntimeMachine { self.pending_effects.push_back(RuntimeEffect::RunWorkflow { workflow_name, changed_files: Vec::new(), + origin: WorkflowRunOrigin::Startup, }); } self.pending_effects.push_back(RuntimeEffect::StartWatching); @@ -384,6 +392,7 @@ impl RuntimeMachine { self.pending_effects.push_back(RuntimeEffect::RunWorkflow { workflow_name, changed_files, + origin: WorkflowRunOrigin::Runtime, }); } } @@ -413,6 +422,7 @@ impl RuntimeMachine { self.pending_effects.push_front(RuntimeEffect::RunWorkflow { workflow_name, changed_files: Vec::new(), + origin: WorkflowRunOrigin::Runtime, }); } RuntimeEvent::CtrlC => { @@ -827,14 +837,16 @@ mod tests { runtime.next_effect(), Some(RuntimeEffect::RunWorkflow { workflow_name: "startup".into(), - changed_files: Vec::new() + changed_files: Vec::new(), + origin: WorkflowRunOrigin::Startup, }) ); assert_eq!( runtime.next_effect(), Some(RuntimeEffect::RunWorkflow { workflow_name: "publish".into(), - changed_files: Vec::new() + changed_files: Vec::new(), + origin: WorkflowRunOrigin::Startup, }) ); assert_eq!(runtime.next_effect(), Some(RuntimeEffect::StartWatching)); @@ -890,7 +902,8 @@ mod tests { runtime.next_effect(), Some(RuntimeEffect::RunWorkflow { workflow_name: "rust".into(), - changed_files: vec!["src/main.rs".into()] + changed_files: vec!["src/main.rs".into()], + origin: WorkflowRunOrigin::Runtime, }) ); assert_eq!( @@ -967,6 +980,7 @@ mod tests { Some(RuntimeEffect::RunWorkflow { workflow_name: "publish_post_url".into(), changed_files: Vec::new(), + origin: WorkflowRunOrigin::Runtime, }) ); assert_eq!(runtime.next_effect(), None); diff --git a/src/engine.rs b/src/engine.rs index d320bd6..93f57f7 100644 --- a/src/engine.rs +++ b/src/engine.rs @@ -4,7 +4,7 @@ use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; use std::time::Duration; -use anyhow::{Result, anyhow}; +use anyhow::{Context, Result, anyhow}; use notify::{ Config as NotifyConfig, Event, EventKind, PollWatcher, RecommendedWatcher, RecursiveMode, Watcher, @@ -18,9 +18,11 @@ use unicode_width::UnicodeWidthStr; use crate::browser_reload::{BrowserReloadSender, BrowserReloadServer, notify_browser_reload}; use crate::config::{CompiledWatchGroup, CompiledWatchTarget, Config, LogStyle, WatcherKind}; -use crate::core::{RuntimeEffect, RuntimeEvent, RuntimeMachine, WorkflowEffect, WorkflowMachine}; +use crate::core::{ + RuntimeEffect, RuntimeEvent, RuntimeMachine, WorkflowEffect, WorkflowMachine, WorkflowRunOrigin, +}; use crate::external_events::{ExternalEventMessage, ExternalEventServer}; -use crate::processes::ProcessManager; +use crate::processes::{ManagedAddressInUse, ProcessManager}; use crate::session_log::SessionLog; use crate::state::SessionState; use devloop::process_guardian::GuardianExecutable; @@ -54,7 +56,12 @@ trait RuntimeEffectAdapter { async fn start_external_event_server(&mut self) -> Result<()>; async fn start_browser_reload_server(&mut self) -> Result<()>; async fn start_autostart_processes(&mut self) -> Result<()>; - async fn run_workflow(&mut self, workflow_name: &str, changed_files: &[String]) -> Result<()>; + async fn run_workflow( + &mut self, + workflow_name: &str, + changed_files: &[String], + origin: WorkflowRunOrigin, + ) -> Result<()>; async fn start_watching(&mut self) -> Result<()>; async fn maintain_processes(&mut self) -> Result<()>; async fn poll_observed_hook(&mut self, hook: &str) -> Result; @@ -67,6 +74,7 @@ struct LiveWorkflowAdapter<'a, 'b> { processes: &'a mut ProcessManager<'b>, state: &'a SessionState, browser_reload_sender: Option, + restart_after_failed_start: bool, } struct LiveRuntimeAdapter<'a, 'b> { @@ -223,7 +231,9 @@ impl WorkflowEffectAdapter for LiveWorkflowAdapter<'_, '_> { } async fn wait_for_process(&mut self, process: &str) -> Result<()> { - self.processes.wait_for_named(process, self.state).await + self.processes + .wait_for_named(process, self.state, self.restart_after_failed_start) + .await } async fn run_hook( @@ -302,7 +312,12 @@ impl RuntimeEffectAdapter for LiveRuntimeAdapter<'_, '_> { self.processes.start_autostart(self.state).await } - async fn run_workflow(&mut self, workflow_name: &str, changed_files: &[String]) -> Result<()> { + async fn run_workflow( + &mut self, + workflow_name: &str, + changed_files: &[String], + origin: WorkflowRunOrigin, + ) -> Result<()> { info!("running workflow {}", workflow_name); let mut adapter = LiveWorkflowAdapter { processes: self.processes, @@ -311,6 +326,7 @@ impl RuntimeEffectAdapter for LiveRuntimeAdapter<'_, '_> { .browser_reload_server .as_ref() .map(BrowserReloadServer::sender), + restart_after_failed_start: origin == WorkflowRunOrigin::Runtime, }; execute_workflow(self.config, &mut adapter, workflow_name, changed_files).await } @@ -409,12 +425,32 @@ async fn execute_runtime_effects( RuntimeEffect::StartBrowserReloadServer => { adapter.start_browser_reload_server().await? } - RuntimeEffect::StartAutostartProcesses => adapter.start_autostart_processes().await?, + RuntimeEffect::StartAutostartProcesses => { + if let Err(error) = adapter.start_autostart_processes().await { + adapter + .stop_all_processes() + .await + .context("failed to clean up after autostart failure")?; + return Err(error); + } + } RuntimeEffect::RunWorkflow { workflow_name, changed_files, + origin, } => { - if let Err(error) = adapter.run_workflow(&workflow_name, &changed_files).await { + if let Err(error) = adapter + .run_workflow(&workflow_name, &changed_files, origin) + .await + { + if origin == WorkflowRunOrigin::Startup + && error.downcast_ref::().is_some() + { + adapter.stop_all_processes().await.with_context(|| { + format!("failed to clean up after startup workflow '{workflow_name}'") + })?; + return Err(error); + } error!( workflow = %workflow_name, error = %workflow_failure_chain(&error), @@ -496,6 +532,7 @@ async fn run_workflow( processes, state, browser_reload_sender, + restart_after_failed_start: true, }; execute_workflow(config, &mut adapter, workflow_name, changed_files).await } @@ -1601,6 +1638,7 @@ mod tests { calls: Vec, changed_hooks: BTreeMap, workflow_errors: BTreeMap, + workflow_address_conflicts: BTreeSet, stop_watching_error: Option, watching: bool, } @@ -1611,6 +1649,7 @@ mod tests { calls: Vec::new(), changed_hooks: BTreeMap::new(), workflow_errors: BTreeMap::new(), + workflow_address_conflicts: BTreeSet::new(), stop_watching_error: None, watching: false, } @@ -1642,7 +1681,11 @@ mod tests { &mut self, workflow_name: &str, changed_files: &[String], + _origin: WorkflowRunOrigin, ) -> Result<()> { + if self.workflow_address_conflicts.contains(workflow_name) { + return Err(ManagedAddressInUse::new("server", "127.0.0.1:8787").into()); + } if let Some(message) = self.workflow_errors.get(workflow_name) { return Err(anyhow!(message.clone())); } @@ -2003,6 +2046,42 @@ mod tests { assert_eq!(adapter.calls, vec!["persist:root", "autostart", "watch"]); } + #[tokio::test] + async fn startup_address_collision_stops_and_cleans_up_before_watching() { + let config = Config { + root: PathBuf::from("."), + debounce_ms: 100, + watcher: crate::config::WatcherConfig::default(), + state_file: Some(PathBuf::from("./state.json")), + startup_workflows: vec!["startup".into()], + watch: BTreeMap::new(), + process: BTreeMap::new(), + hook: BTreeMap::new(), + event_server: crate::config::EventServerConfig::default(), + browser_reload_server: crate::config::BrowserReloadServerConfig::default(), + event: BTreeMap::new(), + workflow: BTreeMap::new(), + }; + let mut runtime = RuntimeMachine::new(&config); + runtime.handle_event(RuntimeEvent::Start { + root_display: "/tmp/example".into(), + startup_workflows: vec!["startup".into()], + }); + let mut adapter = MockRuntimeAdapter::new(); + adapter.workflow_address_conflicts.insert("startup".into()); + + let error = execute_runtime_effects(&mut runtime, &mut adapter) + .await + .expect_err("startup collision must stop the runtime"); + + assert_eq!( + error.to_string(), + "process 'server' cannot start: address 127.0.0.1:8787 is already in use" + ); + assert_eq!(adapter.calls, vec!["persist:root", "autostart", "stop_all"]); + assert!(!adapter.watching); + } + #[tokio::test] async fn missing_runtime_workflow_returns_error() { let config = Config { diff --git a/src/main.rs b/src/main.rs index 518509d..338e86e 100644 --- a/src/main.rs +++ b/src/main.rs @@ -13,6 +13,7 @@ mod test_support; use std::io::{self, Write}; use std::path::PathBuf; +use std::process::ExitCode; use std::time::Duration; use anyhow::{Context, Result, anyhow}; @@ -20,7 +21,7 @@ use clap::{Parser, Subcommand, ValueEnum}; use pulldown_cmark::{ CodeBlockKind, Event as MarkdownEvent, HeadingLevel, Parser as MarkdownParser, Tag, TagEnd, }; -use tracing::{Event, Subscriber, error}; +use tracing::{Event, Subscriber}; use tracing_subscriber::EnvFilter; use tracing_subscriber::fmt::FmtContext; use tracing_subscriber::fmt::MakeWriter; @@ -77,7 +78,17 @@ enum DocsTopic { } #[tokio::main] -async fn main() -> Result<()> { +async fn main() -> ExitCode { + match run_cli().await { + Ok(exit_code) => exit_code, + Err(error) => { + eprintln!("devloop: {error:#}"); + ExitCode::FAILURE + } + } +} + +async fn run_cli() -> Result { let cli = Cli::parse(); match cli.command { Command::Validate { config } => { @@ -102,9 +113,9 @@ async fn main() -> Result<()> { .run() .await { - error!(error = %format!("{error:#}"), "devloop run failed"); + report_run_failure(&session_log, &error); flush_session_log_before_exit(&session_log).await; - return Err(error); + return Ok(ExitCode::FAILURE); } flush_session_log_before_exit(&session_log).await; } @@ -112,7 +123,15 @@ async fn main() -> Result<()> { print!("{}", render_docs_text(topic)); } } - Ok(()) + Ok(ExitCode::SUCCESS) +} + +fn report_run_failure(session_log: &SessionLog, error: &anyhow::Error) { + let message = format!("devloop run failed: {error:#}"); + if let Err(log_error) = session_log.write_labeled_line("devloop", message.as_bytes()) { + eprintln!("devloop: failed to persist run failure: {log_error}"); + } + eprintln!("devloop: {message}"); } async fn flush_session_log_before_exit(session_log: &SessionLog) { diff --git a/src/processes.rs b/src/processes.rs index 3529757..67a80c2 100644 --- a/src/processes.rs +++ b/src/processes.rs @@ -1,4 +1,6 @@ use std::collections::{BTreeMap, VecDeque}; +use std::fmt; +use std::net::{IpAddr, SocketAddr, TcpListener}; use std::os::fd::AsRawFd; use std::os::unix::net::UnixStream; use std::path::{Path, PathBuf}; @@ -81,6 +83,37 @@ const TERMINAL_OUTPUT_QUEUE_CAPACITY: usize = 256; const SESSION_LOG_FLUSH_TIMEOUT: Duration = Duration::from_secs(5); const PROCESS_STOP_TIMEOUT: Duration = Duration::from_secs(2); const GUARDIAN_REAP_TIMEOUT: Duration = Duration::from_secs(2); +#[derive(Debug)] +pub(crate) struct ManagedAddressInUse { + process: String, + address: String, +} + +impl ManagedAddressInUse { + pub(crate) fn new(process: &str, address: impl ToString) -> Self { + Self { + process: process.to_owned(), + address: address.to_string(), + } + } +} + +struct LocalReadinessAddress { + display: String, + candidates: Vec, +} + +impl fmt::Display for ManagedAddressInUse { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + write!( + formatter, + "process '{}' cannot start: address {} is already in use", + self.process, self.address + ) + } +} + +impl std::error::Error for ManagedAddressInUse {} struct CommandContext<'a> { env: &'a BTreeMap, @@ -136,11 +169,19 @@ impl<'a> ProcessManager<'a> { if self.shutting_down { return Ok(()); } + if self.children.contains_key(name) { + return Ok(()); + } let spec = self .config .process .get(name) .ok_or_else(|| anyhow!("unknown process '{name}'"))?; + if let Some(address) = local_readiness_address(name, spec)? + && address_is_occupied(&address) + { + return Err(ManagedAddressInUse::new(name, address.display).into()); + } self.start(name, spec, state).await } @@ -175,7 +216,12 @@ impl<'a> ProcessManager<'a> { self.start_named(name, state).await } - pub async fn wait_for_named(&mut self, name: &str, state: &SessionState) -> Result<()> { + pub async fn wait_for_named( + &mut self, + name: &str, + state: &SessionState, + restart_after_failed_start: bool, + ) -> Result<()> { let (readiness, output_rules) = self .config .process @@ -184,7 +230,7 @@ impl<'a> ProcessManager<'a> { .ok_or_else(|| anyhow!("unknown process '{name}'"))?; let Some(probe) = readiness else { return self - .ensure_process_running(name, &output_rules, state) + .ensure_process_running(name, &output_rules, state, restart_after_failed_start) .await; }; let probe = expand_probe_env(name, &probe)?; @@ -196,10 +242,10 @@ impl<'a> ProcessManager<'a> { }; let interval = Duration::from_millis(probe.interval()); loop { - self.ensure_process_running(name, &output_rules, state) + self.ensure_process_running(name, &output_rules, state, restart_after_failed_start) .await?; if check_probe(&self.client, name, &probe, state).await.is_ok() { - self.ensure_process_running(name, &output_rules, state) + self.ensure_process_running(name, &output_rules, state, restart_after_failed_start) .await?; return Ok(()); } @@ -215,6 +261,7 @@ impl<'a> ProcessManager<'a> { name: &str, output_rules: &[OutputRule], state: &SessionState, + restart_after_failed_start: bool, ) -> Result<()> { let child = self .children @@ -229,13 +276,29 @@ impl<'a> ProcessManager<'a> { self.retire_output_state(name, output_rules, state); signal_process_group(name, managed.guarded.process_group, Signal::KILL)?; self.spawn_output_cleanup(name.to_owned(), managed.output_tasks); - let now_ms = self.clock_start.elapsed().as_millis() as u64; - for effect in self.supervisor.on_tick( - self.config, - now_ms, - vec![(name.to_owned(), status.success())], - ) { - self.apply_process_effect(effect, state).await?; + let occupied_address = if let Some(spec) = self.config.process.get(name) + && let Some(address) = local_readiness_address(name, spec)? + && address_is_occupied(&address) + { + Some(address.display) + } else { + None + }; + if let Some(address) = occupied_address { + self.supervisor.on_process_stopped(name); + return Err(ManagedAddressInUse::new(name, address).into()); + } + if restart_after_failed_start { + let now_ms = self.clock_start.elapsed().as_millis() as u64; + for effect in self.supervisor.on_tick( + self.config, + now_ms, + vec![(name.to_owned(), status.success())], + ) { + self.apply_process_effect(effect, state).await?; + } + } else { + self.supervisor.on_process_stopped(name); } return Err(anyhow!( "process '{name}' exited with {status} before it became ready" @@ -630,6 +693,59 @@ impl<'a> ProcessManager<'a> { } } +fn local_readiness_address( + name: &str, + spec: &ProcessSpec, +) -> Result> { + let Some(readiness) = &spec.readiness else { + return Ok(None); + }; + let ProbeSpec::Http { url, .. } = expand_probe_env(name, readiness)? else { + return Ok(None); + }; + let url = reqwest::Url::parse(&url) + .with_context(|| format!("invalid HTTP readiness URL for process '{name}'"))?; + let port = url + .port_or_known_default() + .ok_or_else(|| anyhow!("HTTP readiness URL for process '{name}' has no port"))?; + let Some(host) = url.host_str() else { + return Ok(None); + }; + if host == "localhost" { + return Ok(Some(LocalReadinessAddress { + display: format!("localhost:{port}"), + candidates: vec![ + SocketAddr::new(IpAddr::V4(std::net::Ipv4Addr::LOCALHOST), port), + SocketAddr::new(IpAddr::V6(std::net::Ipv6Addr::LOCALHOST), port), + ], + })); + } + let host = host.trim_start_matches('[').trim_end_matches(']'); + let ip = if let Ok(ip) = host.parse::() + && ip.is_loopback() + { + ip + } else { + return Ok(None); + }; + let address = SocketAddr::new(ip, port); + Ok(Some(LocalReadinessAddress { + display: address.to_string(), + candidates: vec![address], + })) +} + +fn address_is_occupied(address: &LocalReadinessAddress) -> bool { + for candidate in &address.candidates { + match TcpListener::bind(candidate) { + Ok(listener) => drop(listener), + Err(error) if error.kind() == std::io::ErrorKind::AddrInUse => return true, + Err(_) => {} + } + } + false +} + /// Spawns a managed command behind devloop's internal Rust guardian. /// /// The guardian stays outside the target process group, so ordinary TERM/KILL @@ -1966,7 +2082,7 @@ mod tests { ); let error = manager - .wait_for_named("tunnel", &state) + .wait_for_named("tunnel", &state, false) .await .expect_err("stale state must not imply readiness"); @@ -1994,7 +2110,7 @@ mod tests { timeout_ms: 1000, }), liveness: None, - restart: crate::config::RestartPolicy::Never, + restart: crate::config::RestartPolicy::Always, env: BTreeMap::new(), output: OutputConfig { rules: vec![OutputRule { @@ -2041,7 +2157,7 @@ mod tests { .expect("capture readiness before observing server exit"); let error = manager - .wait_for_named("server", &state) + .wait_for_named("server", &state, false) .await .expect_err("exited process must not become ready"); @@ -2408,7 +2524,7 @@ exec sleep 600 .await .expect("start process"); manager - .wait_for_named("server", &state) + .wait_for_named("server", &state, true) .await .expect("wait for first process readiness"); log.fail_for_test( @@ -2422,7 +2538,7 @@ exec sleep 600 .expect("restart process"); manager - .wait_for_named("server", &state) + .wait_for_named("server", &state, true) .await .expect("wait for restarted process readiness"); assert!(manager.children.contains_key("server")); diff --git a/tests/port_collisions.rs b/tests/port_collisions.rs new file mode 100644 index 0000000..e3ad081 --- /dev/null +++ b/tests/port_collisions.rs @@ -0,0 +1,215 @@ +#![cfg(unix)] + +use std::net::{SocketAddr, TcpListener, TcpStream}; +use std::path::Path; +use std::process::{Child, Command, Stdio}; +use std::time::{Duration, Instant}; + +use rustix::process::{Pid, Signal, kill_process}; +use tempfile::TempDir; + +#[test] +fn separate_sessions_can_run_on_distinct_ports() { + let first = SessionFixture::ephemeral(); + let second = SessionFixture::ephemeral(); + + let mut first_session = DevloopChild::spawn(&first); + let mut second_session = DevloopChild::spawn(&second); + + let first_address = wait_for_reported_listener(&first, &mut first_session.child); + let second_address = wait_for_reported_listener(&second, &mut second_session.child); + assert_ne!(first_address, second_address); + first_session.assert_running(); + second_session.assert_running(); +} + +#[test] +fn startup_fails_once_when_a_managed_address_is_occupied() { + let incumbent = TcpListener::bind("127.0.0.1:0").expect("bind incumbent listener"); + assert_startup_collision(&incumbent); +} + +#[test] +fn startup_detects_an_occupied_ipv6_loopback_address() { + let Ok(incumbent) = TcpListener::bind("[::1]:0") else { + return; + }; + assert_startup_collision(&incumbent); +} + +fn assert_startup_collision(incumbent: &TcpListener) { + let address = incumbent.local_addr().expect("read incumbent address"); + let fixture = SessionFixture::new(address); + + let output = Command::new(env!("CARGO_BIN_EXE_devloop")) + .arg("run") + .arg("--config") + .arg(fixture.config_path()) + .current_dir(fixture.path()) + .env("RUST_LOG", "off") + .output() + .expect("run colliding devloop session"); + + assert!(!output.status.success(), "collision must exit non-zero"); + let stderr = String::from_utf8(output.stderr).expect("stderr is UTF-8"); + let diagnostic = format!("process 'server' cannot start: address {address} is already in use"); + assert_eq!( + stderr.matches(&diagnostic).count(), + 1, + "collision must produce one concise diagnostic; stderr: {stderr}" + ); + assert!(!stderr.contains("started process server"), "{stderr}"); + TcpStream::connect(address).expect("incumbent listener remains reachable"); +} + +fn wait_for_reported_listener(fixture: &SessionFixture, child: &mut Child) -> SocketAddr { + let deadline = Instant::now() + Duration::from_secs(10); + loop { + if let Ok(raw_state) = std::fs::read_to_string(fixture.state_path()) + && let Ok(state) = serde_json::from_str::(&raw_state) + && let Some(port) = state + .get("server_port") + .and_then(serde_json::Value::as_str) + .and_then(|port| port.parse::().ok()) + { + let address = SocketAddr::from(([127, 0, 0, 1], port)); + if TcpStream::connect(address).is_ok() { + return address; + } + } + if let Some(status) = child.try_wait().expect("read devloop status") { + panic!("devloop exited before its ephemeral port became ready: {status}"); + } + assert!( + Instant::now() < deadline, + "timed out waiting for devloop's reported listener" + ); + std::thread::yield_now(); + } +} + +struct SessionFixture { + dir: TempDir, +} + +impl SessionFixture { + fn new(address: SocketAddr) -> Self { + let dir = tempfile::tempdir().expect("create session fixture"); + let fixture = Self { dir }; + fixture.copy_server(); + let config = format!( + r#"root = "." +state_file = "./.devloop/state.json" +startup_workflows = ["startup"] + +[watch.config] +paths = ["devloop.toml"] +workflow = "startup" + +[process.server] +command = ["python3", "server.py", "{port}"] +autostart = false +readiness = {{ kind = "http", url = "http://{address}/", interval_ms = 20, timeout_ms = 5000 }} +restart = "always" + +[workflow.startup] +steps = [ + {{ action = "start_process", process = "server" }}, + {{ action = "wait_for_process", process = "server" }}, +] +"#, + port = address.port(), + ); + std::fs::write(fixture.config_path(), config).expect("write session config"); + fixture + } + + fn ephemeral() -> Self { + let dir = tempfile::tempdir().expect("create session fixture"); + let fixture = Self { dir }; + fixture.copy_server(); + let config = r#"root = "." +state_file = "./.devloop/state.json" +startup_workflows = ["startup"] + +[watch.config] +paths = ["devloop.toml"] +workflow = "startup" + +[process.server] +command = ["python3", "server.py", "0"] +autostart = false +readiness = { kind = "state_key", key = "server_port", interval_ms = 20, timeout_ms = 5000 } +restart = "always" +output = { inherit = false, rules = [{ state_key = "server_port", pattern = "^listening ([0-9]+)$", extract = "regex", capture_group = 1 }] } + +[workflow.startup] +steps = [ + { action = "start_process", process = "server" }, + { action = "wait_for_process", process = "server" }, +] +"#; + std::fs::write(fixture.config_path(), config).expect("write session config"); + fixture + } + + fn copy_server(&self) { + std::fs::copy( + Path::new(env!("CARGO_MANIFEST_DIR")).join("fixtures/port-session/server.py"), + self.path().join("server.py"), + ) + .expect("copy server fixture"); + } + + fn path(&self) -> &Path { + self.dir.path() + } + + fn config_path(&self) -> std::path::PathBuf { + self.path().join("devloop.toml") + } + + fn state_path(&self) -> std::path::PathBuf { + self.path().join(".devloop/state.json") + } +} + +struct DevloopChild { + child: Child, +} + +impl DevloopChild { + fn spawn(fixture: &SessionFixture) -> Self { + let child = Command::new(env!("CARGO_BIN_EXE_devloop")) + .arg("run") + .arg("--config") + .arg(fixture.config_path()) + .current_dir(fixture.path()) + .env("RUST_LOG", "info") + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .expect("spawn devloop session"); + Self { child } + } + + fn assert_running(&mut self) { + assert!( + self.child + .try_wait() + .expect("read devloop status") + .is_none(), + "devloop session exited" + ); + } +} + +impl Drop for DevloopChild { + fn drop(&mut self) { + if self.child.try_wait().ok().flatten().is_none() { + let pid = Pid::from_raw(self.child.id() as i32).expect("devloop pid"); + let _ = kill_process(pid, Signal::INT); + } + let _ = self.child.wait(); + } +} From 5301701626817f0f1ed4b251bed9607c284930bd Mon Sep 17 00:00:00 2001 From: Daniel Vianna <1708810+pasunboneleve@users.noreply.github.com> Date: Tue, 1 Sep 2026 16:51:38 +1000 Subject: [PATCH 2/3] Prepare devloop 0.10.4 Context: The managed-port collision fix is validated and ready for a patch release. Decision: Move the unreleased fix into the 2026-09-01 version section and align Cargo package metadata with release 0.10.4. Alternatives considered: A minor release was unnecessary because the change repairs startup failure handling without adding a new command or configuration contract. Tradeoffs: This release documents only the net collision fix; implementation and test detail remain in the preceding commit and pull request. Architectural impact: No further runtime boundaries change in this commit. It synchronizes the changelog, package version, release-notes input, and intended v0.10.4 tag. --- CHANGELOG.md | 2 ++ Cargo.lock | 2 +- Cargo.toml | 2 +- 3 files changed, 4 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 16cffd8..aa00c7a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,8 @@ All notable changes to `devloop` will be recorded in this file. ## [Unreleased] +## [0.10.4] - 2026-09-01 + ### Fixed - Made startup stop with a clear non-zero error when a managed process's local diff --git a/Cargo.lock b/Cargo.lock index 2f44a52..53e298a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -235,7 +235,7 @@ checksum = "1d07550c9036bf2ae0c684c4297d503f838287c83c53686d05370d0e139ae570" [[package]] name = "devloop" -version = "0.10.3" +version = "0.10.4" dependencies = [ "anyhow", "axum", diff --git a/Cargo.toml b/Cargo.toml index 59c9e60..77a8b98 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "devloop" -version = "0.10.3" +version = "0.10.4" edition = "2024" [dependencies] From 9def924fd2d548c73350d69d4edfaec1d34524e0 Mon Sep 17 00:00:00 2001 From: Daniel Vianna <1708810+pasunboneleve@users.noreply.github.com> Date: Tue, 1 Sep 2026 16:56:50 +1000 Subject: [PATCH 3/3] Make port concurrency fixture self-contained --- fixtures/port-session/server.py | 9 ------ tests/port_collisions.rs | 53 ++++++++++++++++++++------------- 2 files changed, 32 insertions(+), 30 deletions(-) delete mode 100644 fixtures/port-session/server.py diff --git a/fixtures/port-session/server.py b/fixtures/port-session/server.py deleted file mode 100644 index e762c22..0000000 --- a/fixtures/port-session/server.py +++ /dev/null @@ -1,9 +0,0 @@ -import http.server -import sys - - -host = "127.0.0.1" -port = int(sys.argv[1]) -server = http.server.ThreadingHTTPServer((host, port), http.server.SimpleHTTPRequestHandler) -print(f"listening {server.server_address[1]}", flush=True) -server.serve_forever() diff --git a/tests/port_collisions.rs b/tests/port_collisions.rs index e3ad081..d2793ef 100644 --- a/tests/port_collisions.rs +++ b/tests/port_collisions.rs @@ -1,13 +1,30 @@ #![cfg(unix)] +use std::io::Write; use std::net::{SocketAddr, TcpListener, TcpStream}; -use std::path::Path; use std::process::{Child, Command, Stdio}; use std::time::{Duration, Instant}; use rustix::process::{Pid, Signal, kill_process}; use tempfile::TempDir; +#[test] +#[ignore] +fn serve_ephemeral_port() { + if std::env::var_os("DEVLOOP_TEST_SERVER_MODE").is_none() { + return; + } + let listener = TcpListener::bind("127.0.0.1:0").expect("bind ephemeral test server"); + println!( + "listening {}", + listener.local_addr().expect("read server address").port() + ); + std::io::stdout().flush().expect("flush server address"); + loop { + let _ = listener.accept().expect("accept test connection"); + } +} + #[test] fn separate_sessions_can_run_on_distinct_ports() { let first = SessionFixture::ephemeral(); @@ -96,7 +113,6 @@ impl SessionFixture { fn new(address: SocketAddr) -> Self { let dir = tempfile::tempdir().expect("create session fixture"); let fixture = Self { dir }; - fixture.copy_server(); let config = format!( r#"root = "." state_file = "./.devloop/state.json" @@ -107,7 +123,7 @@ paths = ["devloop.toml"] workflow = "startup" [process.server] -command = ["python3", "server.py", "{port}"] +command = ["devloop-test-command-that-must-not-run"] autostart = false readiness = {{ kind = "http", url = "http://{address}/", interval_ms = 20, timeout_ms = 5000 }} restart = "always" @@ -118,7 +134,6 @@ steps = [ {{ action = "wait_for_process", process = "server" }}, ] "#, - port = address.port(), ); std::fs::write(fixture.config_path(), config).expect("write session config"); fixture @@ -127,8 +142,9 @@ steps = [ fn ephemeral() -> Self { let dir = tempfile::tempdir().expect("create session fixture"); let fixture = Self { dir }; - fixture.copy_server(); - let config = r#"root = "." + let test_binary = std::env::current_exe().expect("resolve integration test executable"); + let config = format!( + r#"root = "." state_file = "./.devloop/state.json" startup_workflows = ["startup"] @@ -137,31 +153,26 @@ paths = ["devloop.toml"] workflow = "startup" [process.server] -command = ["python3", "server.py", "0"] +command = ["{test_binary}", "--exact", "serve_ephemeral_port", "--ignored", "--nocapture"] autostart = false -readiness = { kind = "state_key", key = "server_port", interval_ms = 20, timeout_ms = 5000 } +readiness = {{ kind = "state_key", key = "server_port", interval_ms = 20, timeout_ms = 5000 }} restart = "always" -output = { inherit = false, rules = [{ state_key = "server_port", pattern = "^listening ([0-9]+)$", extract = "regex", capture_group = 1 }] } +env = {{ DEVLOOP_TEST_SERVER_MODE = "1" }} +output = {{ inherit = false, rules = [{{ state_key = "server_port", pattern = "^listening ([0-9]+)$", extract = "regex", capture_group = 1 }}] }} [workflow.startup] steps = [ - { action = "start_process", process = "server" }, - { action = "wait_for_process", process = "server" }, + {{ action = "start_process", process = "server" }}, + {{ action = "wait_for_process", process = "server" }}, ] -"#; +"#, + test_binary = test_binary.display() + ); std::fs::write(fixture.config_path(), config).expect("write session config"); fixture } - fn copy_server(&self) { - std::fs::copy( - Path::new(env!("CARGO_MANIFEST_DIR")).join("fixtures/port-session/server.py"), - self.path().join("server.py"), - ) - .expect("copy server fixture"); - } - - fn path(&self) -> &Path { + fn path(&self) -> &std::path::Path { self.dir.path() }