ic_testkit/pic/

startup.rs

use std::{
    fmt::Write as _,
    fs::{self, File, OpenOptions},
    io::{self, Read as _},
    path::{Path, PathBuf},
    process::{Child, Command, ExitStatus, Stdio},
    sync::{
        atomic::{AtomicU64, Ordering},
        mpsc::{self, RecvTimeoutError},
    },
    thread,
    time::{Duration, Instant},
};

#[cfg(unix)]
use std::os::unix::{
    fs::{DirBuilderExt as _, OpenOptionsExt as _},
    process::CommandExt as _,
};

use pocket_ic::{PocketIc, PocketIcBuilder};

use super::transport;

const STARTUP_POLL_INTERVAL: Duration = Duration::from_millis(20);
const SERVER_OUTPUT_LIMIT: usize = 16 * 1024;
// PocketIC publishes a decimal u16 and a newline. Leave whitespace room
// without allowing readiness polling to read an arbitrary-size file.
const SERVER_PORT_FILE_LIMIT: usize = 64;

static STARTUP_FILE_SEQUENCE: AtomicU64 = AtomicU64::new(0);

/// Explicit bounded source and policy for one PocketIC startup.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct PocketIcStartupConfig {
    source: PocketIcStartupSource,
    timeout: Duration,
    server_hard_ttl: Option<Duration>,
}

/// Caller-owned PocketIC server process with bounded startup and output capture.
///
/// Dropping the handle terminates and waits for the managed child. On Unix,
/// teardown also terminates descendants remaining in its owned process group.
/// Callers may create several instances through [`Self::url`] and
/// [`PocketIcStartupConfig::connect`] while retaining explicit server ownership.
/// The handle is process-local and does not coordinate ownership across Cargo
/// or test-runner processes; use an externally owned server with bounded
/// connect mode for that topology.
/// The handle owns no binary discovery, download, cache, or compatibility policy.
pub struct PocketIcManagedServer {
    server: ManagedServer,
    url: String,
}

/// Bounded lossy UTF-8 output captured from a managed PocketIC server.
///
/// Each stream retains at most the first 16 KiB. A textual suffix reports the
/// number of omitted bytes when truncation occurred. Unreadable streams and
/// paths replaced with non-regular files are omitted.
#[derive(Clone, Debug, Default, Eq, PartialEq)]
pub struct PocketIcManagedServerOutput {
    stdout: String,
    stderr: String,
}

#[derive(Clone, Debug, Eq, PartialEq)]
enum PocketIcStartupSource {
    Spawn { server_binary: PathBuf },
    Connect { server_url: String },
}

/// Structured failure from bounded PocketIC construction.
#[non_exhaustive]
#[derive(Debug)]
pub enum PocketIcStartupError {
    /// The caller supplied a zero timeout or unusable hard TTL.
    InvalidConfiguration { message: String },
    /// A caller-provided existing server URL could not be parsed.
    InvalidServerUrl { server_url: String, message: String },
    /// Preparing or inspecting bounded startup files failed.
    Io {
        operation: &'static str,
        path: PathBuf,
        source: io::Error,
    },
    /// The configured PocketIC server process could not be spawned.
    ServerSpawn {
        server_binary: PathBuf,
        source: io::Error,
    },
    /// The managed PocketIC server exited before instance construction completed.
    ServerExited {
        server_binary: PathBuf,
        status: ExitStatus,
        elapsed: Duration,
        stdout: String,
        stderr: String,
    },
    /// The managed server did not publish a usable port before the deadline.
    ReadinessTimeout {
        server_binary: PathBuf,
        timeout: Duration,
        stdout: String,
        stderr: String,
        termination_error: Option<String>,
    },
    /// The managed server published an invalid or oversized port-file value.
    InvalidServerPort {
        server_binary: PathBuf,
        value: String,
        stdout: String,
        stderr: String,
    },
    /// PocketIC instance creation did not finish before the startup deadline.
    InstanceCreationTimeout {
        timeout: Duration,
        stdout: String,
        stderr: String,
        termination_error: Option<String>,
    },
    /// Spawning the bounded builder worker failed.
    BuilderThreadSpawn { source: io::Error },
    /// Upstream PocketIC construction panicked before returning an instance.
    BuilderPanicked { message: String },
    /// The bounded builder worker ended without returning a result.
    BuilderDisconnected,
}

/// Fallible construction at PocketIC's panicking builder boundary.
///
/// Startup is explicit: callers either provide an existing server URL or let
/// `ic-testkit` spawn and monitor one exact server binary. This prevents the
/// upstream builder from hiding an unobservable child process.
pub trait PocketIcBuilderExt {
    /// Build one PocketIC instance within the configured deadline.
    ///
    /// Managed server startup detects child exit while awaiting the port file,
    /// terminates the child on timeout, and reads bounded stdout/stderr prefixes.
    /// Instance creation is also bounded. Upstream panics remain structured.
    ///
    /// This deadline covers construction only. Dropping the returned instance
    /// uses PocketIC's synchronous HTTP deletion, which has no request deadline
    /// in PocketIC 16. An operation's maximum request time does not bound drop.
    fn try_build(self, config: PocketIcStartupConfig) -> Result<PocketIc, PocketIcStartupError>;
}

impl PocketIcStartupConfig {
    /// Spawn and monitor one exact PocketIC server binary.
    ///
    /// Startup allocates a unique private temporary directory while leaving
    /// the `--port-file` path absent for PocketIC to create.
    #[must_use]
    pub fn spawn(server_binary: impl Into<PathBuf>, timeout: Duration) -> Self {
        Self {
            source: PocketIcStartupSource::Spawn {
                server_binary: server_binary.into(),
            },
            timeout,
            server_hard_ttl: None,
        }
    }

    /// Connect to a caller-owned existing PocketIC server.
    ///
    /// The URL is applied to the builder explicitly, so this mode never lets
    /// the upstream builder spawn a hidden server child.
    #[must_use]
    pub fn connect(server_url: impl Into<String>, timeout: Duration) -> Self {
        Self {
            source: PocketIcStartupSource::Connect {
                server_url: server_url.into(),
            },
            timeout,
            server_hard_ttl: None,
        }
    }

    /// Set the hard lifetime passed to an `ic-testkit`-managed server.
    #[must_use]
    pub const fn with_server_hard_ttl(mut self, hard_ttl: Duration) -> Self {
        self.server_hard_ttl = Some(hard_ttl);
        self
    }

    /// Complete startup deadline.
    #[must_use]
    pub const fn timeout(&self) -> Duration {
        self.timeout
    }

    /// Explicit managed server hard lifetime, or `None` when disabled.
    #[must_use]
    pub const fn server_hard_ttl(&self) -> Option<Duration> {
        self.server_hard_ttl
    }

    /// Managed server binary, when this configuration spawns one.
    #[must_use]
    pub fn server_binary(&self) -> Option<&Path> {
        match &self.source {
            PocketIcStartupSource::Spawn { server_binary } => Some(server_binary),
            PocketIcStartupSource::Connect { .. } => None,
        }
    }

    /// Existing caller-owned server URL, when configured.
    #[must_use]
    pub fn server_url(&self) -> Option<&str> {
        match &self.source {
            PocketIcStartupSource::Connect { server_url } => Some(server_url),
            PocketIcStartupSource::Spawn { .. } => None,
        }
    }

    /// Start a caller-owned managed server without constructing an instance.
    ///
    /// This requires a configuration created by [`Self::spawn`]. Readiness is
    /// bounded by [`Self::timeout`]. No hard TTL is passed by default; an
    /// explicit [`Self::with_server_hard_ttl`] value is passed to the child.
    /// Readiness requires a nonzero decimal port followed by a newline in a
    /// regular UTF-8 file of at most 64 bytes; oversized files fail with bounded
    /// diagnostics. Non-regular port files fail with [`PocketIcStartupError::Io`]
    /// and [`io::ErrorKind::InvalidData`] without waiting for a FIFO writer.
    /// The returned handle terminates the child on drop; use its URL with
    /// [`Self::connect`] to construct bounded instances.
    pub fn start_managed_server(self) -> Result<PocketIcManagedServer, PocketIcStartupError> {
        self.validate()?;
        let PocketIcStartupSource::Spawn { server_binary } = self.source else {
            return Err(PocketIcStartupError::InvalidConfiguration {
                message: "starting a managed PocketIC server requires a spawn configuration"
                    .to_owned(),
            });
        };
        let started = Instant::now();
        let deadline = startup_deadline(started, self.timeout)?;
        let (server, url) = ManagedServer::start(
            server_binary,
            self.server_hard_ttl,
            deadline,
            self.timeout,
            started,
        )?;
        Ok(PocketIcManagedServer { server, url })
    }

    fn validate(&self) -> Result<(), PocketIcStartupError> {
        if self.timeout.is_zero() {
            return Err(PocketIcStartupError::InvalidConfiguration {
                message: "PocketIC startup timeout must be greater than zero".to_owned(),
            });
        }
        if matches!(&self.source, PocketIcStartupSource::Spawn { .. })
            && self
                .server_hard_ttl
                .is_some_and(|hard_ttl| hard_ttl.as_secs() == 0)
        {
            return Err(PocketIcStartupError::InvalidConfiguration {
                message: "PocketIC server hard TTL must be at least one second".to_owned(),
            });
        }
        Ok(())
    }
}

impl PocketIcManagedServer {
    /// OS process ID of the owned server child, for caller-managed monitoring.
    ///
    /// This identifies the server process, not its descendants, and does not
    /// establish that it is still running. The OS may reuse the ID after the
    /// child exits and is reaped. Dropping this handle terminates and waits for
    /// the child; retaining the ID does not retain server ownership.
    #[must_use]
    pub fn process_id(&self) -> u32 {
        self.server
            .child
            .as_ref()
            .expect("managed server handle must own its child")
            .id()
    }

    /// Loopback URL published by the managed server.
    #[must_use]
    pub fn url(&self) -> &str {
        &self.url
    }

    /// Current bounded stdout and stderr captured from the managed server.
    ///
    /// This reads at most the first 16 KiB from each retained output file,
    /// renders it as lossy UTF-8, and adds an omitted-byte suffix when truncated.
    /// The diagnostic read is bounded; files can grow while the server runs.
    #[must_use]
    pub fn output(&self) -> PocketIcManagedServerOutput {
        self.server.capture().into()
    }
}

impl PocketIcManagedServerOutput {
    /// Bounded lossy UTF-8 standard output.
    #[must_use]
    pub fn stdout(&self) -> &str {
        &self.stdout
    }

    /// Bounded lossy UTF-8 standard error.
    #[must_use]
    pub fn stderr(&self) -> &str {
        &self.stderr
    }
}

impl PocketIcBuilderExt for PocketIcBuilder {
    fn try_build(self, config: PocketIcStartupConfig) -> Result<PocketIc, PocketIcStartupError> {
        config.validate()?;
        let started = Instant::now();
        let deadline = startup_deadline(started, config.timeout)?;
        match config.source {
            PocketIcStartupSource::Connect { server_url } => {
                build_bounded(self, &server_url, deadline, config.timeout, None)
            }
            PocketIcStartupSource::Spawn { server_binary } => {
                let (server, server_url) = ManagedServer::start(
                    server_binary,
                    config.server_hard_ttl,
                    deadline,
                    config.timeout,
                    started,
                )?;
                build_bounded(self, &server_url, deadline, config.timeout, Some(server))
            }
        }
    }
}

fn startup_deadline(started: Instant, timeout: Duration) -> Result<Instant, PocketIcStartupError> {
    started
        .checked_add(timeout)
        .ok_or_else(|| PocketIcStartupError::InvalidConfiguration {
            message: "PocketIC startup timeout exceeds the platform clock range".to_owned(),
        })
}

fn build_bounded(
    builder: PocketIcBuilder,
    server_url: &str,
    deadline: Instant,
    timeout: Duration,
    mut server: Option<ManagedServer>,
) -> Result<PocketIc, PocketIcStartupError> {
    let builder = match server_url.parse() {
        Ok(server_url) => builder.with_server_url(server_url),
        Err(error) => {
            return Err(PocketIcStartupError::InvalidServerUrl {
                server_url: server_url.to_owned(),
                message: error.to_string(),
            });
        }
    };
    let (sender, receiver) = mpsc::sync_channel(1);
    if let Err(source) = thread::Builder::new()
        .name("ic-testkit-pocket-ic-startup".to_owned())
        .spawn(move || {
            let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| builder.build()))
                .map_err(|payload| transport::panic_payload_to_string(payload.as_ref()));
            let _ = sender.send(result);
        })
    {
        return Err(PocketIcStartupError::BuilderThreadSpawn { source });
    }

    loop {
        let now = Instant::now();
        if now >= deadline {
            let captured = server.take().map_or_else(
                CapturedServer::default,
                ManagedServer::terminate_and_capture,
            );
            return Err(PocketIcStartupError::InstanceCreationTimeout {
                timeout,
                stdout: captured.stdout,
                stderr: captured.stderr,
                termination_error: captured.termination_error,
            });
        }
        let remaining = deadline.saturating_duration_since(now);
        let wait = if server.is_some() {
            remaining.min(STARTUP_POLL_INTERVAL)
        } else {
            remaining
        };
        match receiver.recv_timeout(wait) {
            Ok(Ok(pocket_ic)) => {
                if let Some(mut managed) = server.take() {
                    if let Some(status) = managed.try_wait()? {
                        return Err(managed.exited_error(status));
                    }
                    managed.reap_in_background();
                }
                return Ok(pocket_ic);
            }
            Ok(Err(message)) => {
                if let Some(server) = server.take() {
                    let _ = server.terminate_and_capture();
                }
                return Err(PocketIcStartupError::BuilderPanicked { message });
            }
            Err(RecvTimeoutError::Disconnected) => {
                if let Some(server) = server.take() {
                    let _ = server.terminate_and_capture();
                }
                return Err(PocketIcStartupError::BuilderDisconnected);
            }
            Err(RecvTimeoutError::Timeout) => {
                if let Some(managed) = &mut server
                    && let Some(status) = managed.try_wait()?
                {
                    return Err(server
                        .take()
                        .expect("managed server must remain present")
                        .exited_error(status));
                }
            }
        }
    }
}

struct ManagedServer {
    child: Option<Child>,
    binary: PathBuf,
    files: StartupFiles,
    started: Instant,
}

enum PortFileState {
    Pending,
    Ready(u16),
    Invalid(String),
}

impl ManagedServer {
    fn start(
        binary: PathBuf,
        hard_ttl: Option<Duration>,
        deadline: Instant,
        timeout: Duration,
        started: Instant,
    ) -> Result<(Self, String), PocketIcStartupError> {
        let (files, stdout, stderr) = StartupFiles::create()?;
        let mut command = Command::new(&binary);
        if let Some(hard_ttl) = hard_ttl {
            command
                .arg("--hard-ttl")
                .arg(hard_ttl.as_secs().to_string());
        }
        command
            .arg("--port-file")
            .arg(&files.port)
            .stdout(Stdio::from(stdout))
            .stderr(Stdio::from(stderr));
        #[cfg(unix)]
        {
            command.process_group(0);
        }
        let child = command
            .spawn()
            .map_err(|source| PocketIcStartupError::ServerSpawn {
                server_binary: binary.clone(),
                source,
            })?;
        let mut server = Self {
            child: Some(child),
            binary,
            files,
            started,
        };

        loop {
            if let Some(status) = server.try_wait()? {
                return Err(server.exited_error(status));
            }
            let now = Instant::now();
            if now >= deadline {
                let binary = server.binary.clone();
                let captured = server.terminate_and_capture();
                return Err(PocketIcStartupError::ReadinessTimeout {
                    server_binary: binary,
                    timeout,
                    stdout: captured.stdout,
                    stderr: captured.stderr,
                    termination_error: captured.termination_error,
                });
            }
            match server.read_port()? {
                PortFileState::Pending => {}
                PortFileState::Ready(port) => {
                    return Ok((server, format!("http://127.0.0.1:{port}/")));
                }
                PortFileState::Invalid(value) => {
                    let binary = server.binary.clone();
                    let captured = server.terminate_and_capture();
                    return Err(PocketIcStartupError::InvalidServerPort {
                        server_binary: binary,
                        value,
                        stdout: captured.stdout,
                        stderr: captured.stderr,
                    });
                }
            }
            thread::sleep(
                deadline
                    .saturating_duration_since(now)
                    .min(STARTUP_POLL_INTERVAL),
            );
        }
    }

    fn try_wait(&mut self) -> Result<Option<ExitStatus>, PocketIcStartupError> {
        let child = self
            .child
            .as_mut()
            .expect("managed server child must remain present");
        poll_child(child).map_err(|source| PocketIcStartupError::Io {
            operation: "inspect PocketIC server child",
            path: self.binary.clone(),
            source,
        })
    }

    fn read_port(&self) -> Result<PortFileState, PocketIcStartupError> {
        let port_path = &self.files.port;
        let mut contents = String::new();
        match open_regular_startup_file(port_path).and_then(|file| {
            file.take((SERVER_PORT_FILE_LIMIT + 1) as u64)
                .read_to_string(&mut contents)
        }) {
            Ok(_) => {}
            Err(error) if error.kind() == io::ErrorKind::NotFound => {
                return Ok(PortFileState::Pending);
            }
            Err(source) => {
                return Err(PocketIcStartupError::Io {
                    operation: "read PocketIC server port file",
                    path: port_path.clone(),
                    source,
                });
            }
        }
        if contents.len() > SERVER_PORT_FILE_LIMIT {
            return Ok(PortFileState::Invalid(format!(
                "{} (port file exceeds {SERVER_PORT_FILE_LIMIT} bytes)",
                contents.trim()
            )));
        }
        if !contents.contains('\n') {
            return Ok(PortFileState::Pending);
        }
        let value = contents.trim().to_owned();
        match value.parse::<u16>() {
            Ok(port) if port != 0 => Ok(PortFileState::Ready(port)),
            _ => Ok(PortFileState::Invalid(value)),
        }
    }

    fn exited_error(mut self, status: ExitStatus) -> PocketIcStartupError {
        let elapsed = self.started.elapsed();
        let binary = self.binary.clone();
        self.child.take();
        let captured = self.capture();
        PocketIcStartupError::ServerExited {
            server_binary: binary,
            status,
            elapsed,
            stdout: captured.stdout,
            stderr: captured.stderr,
        }
    }

    fn terminate_and_capture(mut self) -> CapturedServer {
        let termination_error = match self.child.take() {
            Some(mut child) => terminate_child(&mut child),
            None => None,
        };
        let mut captured = self.capture();
        captured.termination_error = termination_error;
        captured
    }

    fn capture(&self) -> CapturedServer {
        let files = &self.files;
        CapturedServer {
            stdout: read_bounded_lossy(&files.stdout),
            stderr: read_bounded_lossy(&files.stderr),
            termination_error: None,
        }
    }

    fn reap_in_background(self) {
        let _ = thread::Builder::new()
            .name("ic-testkit-pocket-ic-server-reaper".to_owned())
            .spawn(move || {
                // Keep child and files under one owner, including if spawning
                // this thread fails. On Unix, Drop terminates the group before reaping.
                let mut server = self;
                if let Some(child) = server.child.as_mut() {
                    #[cfg(unix)]
                    let _ = wait_for_child_exit(child, false);
                    #[cfg(not(unix))]
                    let _ = child.wait();
                }
            });
    }
}

impl Drop for ManagedServer {
    fn drop(&mut self) {
        if let Some(mut child) = self.child.take() {
            let _ = terminate_child(&mut child);
        }
    }
}

#[cfg(unix)]
fn wait_for_child_exit(child: &Child, nonblocking: bool) -> io::Result<bool> {
    // WNOWAIT keeps the leader's PID reserved until group cleanup completes.
    let flags = libc::WEXITED | libc::WNOWAIT | if nonblocking { libc::WNOHANG } else { 0 };
    loop {
        // SAFETY: siginfo_t is a C value for which zero initialization is valid.
        let mut info: libc::siginfo_t = unsafe { std::mem::zeroed() };
        // SAFETY: the PID is an owned, unreaped child and info is writable.
        let result = unsafe { libc::waitid(libc::P_PID, child.id(), &raw mut info, flags) };
        if result == 0 {
            return Ok(info.si_signo != 0);
        }
        let error = io::Error::last_os_error();
        if error.kind() != io::ErrorKind::Interrupted {
            return Err(error);
        }
    }
}

fn poll_child(child: &mut Child) -> io::Result<Option<ExitStatus>> {
    #[cfg(unix)]
    {
        if !wait_for_child_exit(child, true)? {
            return Ok(None);
        }
        // The leader is still unreaped, so its process group cannot be reused.
        terminate_process_group(child)?;
        child.wait().map(Some)
    }
    #[cfg(not(unix))]
    child.try_wait()
}

#[cfg(unix)]
fn terminate_process_group(child: &Child) -> io::Result<()> {
    let group = libc::pid_t::try_from(child.id())
        .map_err(|_| io::Error::other("managed child PID exceeds the OS process ID range"))?;
    loop {
        // SAFETY: spawn sets this child's PGID to its PID, and it remains
        // unreaped until after this call. No caller/test-runner group is used.
        let result = unsafe { libc::killpg(group, libc::SIGKILL) };
        if result == 0 {
            return Ok(());
        }
        let error = io::Error::last_os_error();
        if error.raw_os_error() == Some(libc::ESRCH) {
            return Ok(());
        }
        #[cfg(target_os = "macos")]
        if error.raw_os_error() == Some(libc::EPERM) && wait_for_child_exit(child, true)? {
            // Darwin's group signal path excludes zombies and can report EPERM
            // for the unreaped leader alone. Verify that exact state; permission
            // failures for a group with any other member still propagate.
            // Two slots distinguish the sole leader from a larger/truncated group.
            let mut members: [libc::pid_t; 2] = [0; 2];
            let size = libc::c_int::try_from(std::mem::size_of_val(&members))
                .expect("two process IDs fit in a libproc buffer size");
            // SAFETY: members is writable for size bytes. The owned unreaped
            // leader reserves this group ID across the query and later wait.
            let count =
                unsafe { libc::proc_listpgrppids(group, members.as_mut_ptr().cast(), size) };
            if count == 1 && members[0] == group {
                return Ok(());
            }
        }
        if error.kind() != io::ErrorKind::Interrupted {
            return Err(error);
        }
    }
}

#[cfg(unix)]
fn terminate_child(child: &mut Child) -> Option<String> {
    let termination = terminate_process_group(child);
    if termination.is_err() {
        let _ = child.kill();
    }
    let wait = child.wait();
    termination
        .and_then(|()| wait.map(|_| ()))
        .err()
        .map(|error| error.to_string())
}

#[cfg(not(unix))]
fn terminate_child(child: &mut Child) -> Option<String> {
    match child.try_wait() {
        Ok(Some(_)) => None,
        Ok(None) => child
            .kill()
            .and_then(|()| child.wait().map(|_| ()))
            .err()
            .map(|error| error.to_string()),
        Err(inspect_error) => {
            let termination_error = child.kill().and_then(|()| child.wait().map(|_| ())).err();
            termination_error.map(|termination_error| {
                format!(
                    "failed to inspect child before termination: {inspect_error}; termination also failed: {termination_error}"
                )
            })
        }
    }
}

#[derive(Default)]
struct CapturedServer {
    stdout: String,
    stderr: String,
    termination_error: Option<String>,
}

impl From<CapturedServer> for PocketIcManagedServerOutput {
    fn from(captured: CapturedServer) -> Self {
        Self {
            stdout: captured.stdout,
            stderr: captured.stderr,
        }
    }
}

struct StartupFiles {
    directory: PathBuf,
    port: PathBuf,
    stdout: PathBuf,
    stderr: PathBuf,
}

impl StartupFiles {
    fn create() -> Result<(Self, File, File), PocketIcStartupError> {
        loop {
            let sequence = STARTUP_FILE_SEQUENCE.fetch_add(1, Ordering::Relaxed);
            let base = std::env::temp_dir().join(format!(
                "ic-testkit-pocket-ic-startup-{}-{sequence}",
                std::process::id()
            ));
            let mut directory = fs::DirBuilder::new();
            #[cfg(unix)]
            {
                directory.mode(0o700);
            }
            match directory.create(&base) {
                Ok(()) => {}
                Err(error) if error.kind() == io::ErrorKind::AlreadyExists => continue,
                Err(source) => return Err(startup_file_error("create", &base, source)),
            }
            let files = Self {
                port: base.join("port"),
                stdout: base.join("stdout"),
                stderr: base.join("stderr"),
                directory: base,
            };
            let stdout = create_new_file(&files.stdout)
                .map_err(|source| startup_file_error("create", &files.stdout, source))?;
            let stderr = create_new_file(&files.stderr)
                .map_err(|source| startup_file_error("create", &files.stderr, source))?;
            return Ok((files, stdout, stderr));
        }
    }
}

impl Drop for StartupFiles {
    fn drop(&mut self) {
        let _ = fs::remove_dir_all(&self.directory);
    }
}

fn create_new_file(path: &Path) -> io::Result<File> {
    OpenOptions::new().write(true).create_new(true).open(path)
}

fn startup_file_error(
    operation: &'static str,
    path: &Path,
    source: io::Error,
) -> PocketIcStartupError {
    PocketIcStartupError::Io {
        operation,
        path: path.to_owned(),
        source,
    }
}

fn read_bounded_lossy(path: &Path) -> String {
    let Ok(file) = open_regular_startup_file(path) else {
        return String::new();
    };
    let length = file.metadata().map_or(0, |metadata| metadata.len());
    let mut bytes = Vec::with_capacity(SERVER_OUTPUT_LIMIT);
    if file
        .take(SERVER_OUTPUT_LIMIT as u64)
        .read_to_end(&mut bytes)
        .is_err()
    {
        return String::new();
    }
    let mut output = String::from_utf8_lossy(&bytes).into_owned();
    let omitted = length.saturating_sub(bytes.len() as u64);
    if omitted > 0 {
        let _ = write!(output, "\n<truncated {omitted} bytes>");
    }
    output
}

fn open_regular_startup_file(path: &Path) -> io::Result<File> {
    let mut options = OpenOptions::new();
    options.read(true);
    // Bound opening a replaced FIFO as well as reading file contents. Inspect
    // the opened file, rather than a path that can change before open completes.
    #[cfg(unix)]
    options.custom_flags(libc::O_NONBLOCK);
    let file = options.open(path)?;
    if !file.metadata()?.is_file() {
        return Err(io::Error::new(
            io::ErrorKind::InvalidData,
            "PocketIC startup reader requires a regular file",
        ));
    }
    Ok(file)
}

impl std::fmt::Display for PocketIcStartupError {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::InvalidConfiguration { message } => formatter.write_str(message),
            Self::InvalidServerUrl {
                server_url,
                message,
            } => write!(
                formatter,
                "invalid PocketIC server URL {server_url:?}: {message}"
            ),
            Self::Io {
                operation,
                path,
                source,
            } => write!(
                formatter,
                "failed to {operation} at {}: {source}",
                path.display()
            ),
            Self::ServerSpawn {
                server_binary,
                source,
            } => write!(
                formatter,
                "failed to spawn PocketIC server {}: {source}",
                server_binary.display()
            ),
            Self::ServerExited {
                server_binary,
                status,
                elapsed,
                stderr,
                ..
            } => write!(
                formatter,
                "PocketIC server {} exited with {status} after {elapsed:?}: {stderr}",
                server_binary.display()
            ),
            Self::ReadinessTimeout {
                server_binary,
                timeout,
                ..
            } => write!(
                formatter,
                "PocketIC server {} was not ready within {timeout:?}",
                server_binary.display()
            ),
            Self::InvalidServerPort {
                server_binary,
                value,
                ..
            } => write!(
                formatter,
                "PocketIC server {} published invalid port {value:?}",
                server_binary.display()
            ),
            Self::InstanceCreationTimeout { timeout, .. } => {
                write!(formatter, "PocketIC instance creation exceeded {timeout:?}")
            }
            Self::BuilderThreadSpawn { source } => {
                write!(
                    formatter,
                    "failed to spawn PocketIC builder worker: {source}"
                )
            }
            Self::BuilderPanicked { message } => {
                write!(formatter, "PocketIC startup panicked: {message}")
            }
            Self::BuilderDisconnected => {
                formatter.write_str("PocketIC builder worker disconnected without a result")
            }
        }
    }
}

impl std::error::Error for PocketIcStartupError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            Self::Io { source, .. }
            | Self::ServerSpawn { source, .. }
            | Self::BuilderThreadSpawn { source } => Some(source),
            _ => None,
        }
    }
}

#[cfg(test)]
mod tests {
    use std::{
        fs,
        path::PathBuf,
        time::{Duration, Instant},
    };

    use super::{
        PocketIcBuilderExt as _, PocketIcStartupConfig, PocketIcStartupError, StartupFiles,
    };
    use pocket_ic::PocketIcBuilder;

    #[cfg(unix)]
    use crate::test_executable::write_executable_script;
    #[cfg(unix)]
    use std::{
        io::Write as _,
        os::unix::fs::{OpenOptionsExt as _, PermissionsExt as _},
        process::Command,
        sync::mpsc,
    };

    #[cfg(unix)]
    fn process_state(pid: u32) -> Option<char> {
        // Both supported Unix hosts provide this ps field. A zombie has stopped
        // running but may remain visible until its parent reaps it.
        let output = Command::new("/bin/ps")
            .args(["-p", &pid.to_string(), "-o", "stat="])
            .output()
            .expect("inspect managed test process state");
        assert!(
            output.status.success()
                || (output.status.code() == Some(1)
                    && output.stdout.is_empty()
                    && output.stderr.is_empty()),
            "process-state inspection failed: {}: {}",
            output.status,
            String::from_utf8_lossy(&output.stderr),
        );
        String::from_utf8(output.stdout)
            .expect("process state is ASCII")
            .trim()
            .chars()
            .next()
    }

    #[cfg(unix)]
    #[test]
    fn reading_large_sparse_server_output_is_bounded() {
        let (files, _, _) = StartupFiles::create().expect("allocate startup files");
        let mut file = fs::File::create(&files.stdout).expect("create sparse log");
        file.write_all(b"server started\n").expect("write prefix");
        let size = 8_u64 * 1024 * 1024 * 1024;
        file.set_len(size).expect("extend sparse log");
        let output = super::read_bounded_lossy(&files.stdout);
        assert!(output.starts_with("server started\n"));
        assert!(output.ends_with(&format!(
            "<truncated {} bytes>",
            size - super::SERVER_OUTPUT_LIMIT as u64
        )));
        assert!(output.len() < super::SERVER_OUTPUT_LIMIT + 100);
    }

    #[cfg(unix)]
    #[test]
    fn startup_readers_reject_fifos_without_waiting_for_a_writer() {
        let (files, stdout, stderr) = StartupFiles::create().expect("allocate startup files");
        drop((stdout, stderr));
        fs::remove_file(&files.stdout).unwrap();
        fs::remove_file(&files.stderr).unwrap();
        assert!(
            Command::new("mkfifo")
                .args([&files.port, &files.stdout, &files.stderr])
                .status()
                .expect("create FIFO startup files")
                .success()
        );
        let server = super::ManagedServer {
            child: None,
            binary: PathBuf::from("unused-server"),
            files,
            started: Instant::now(),
        };
        for path in [
            &server.files.port,
            &server.files.stdout,
            &server.files.stderr,
        ] {
            // A delayed writer bounds a blocked read and records whether the
            // reader needed it. Completion cancels the writer; with no reader,
            // a nonblocking open fails and is retried if the reader starts late.
            let fifo = path.clone();
            let (stop_writer, stopped) = mpsc::channel();
            let writer = std::thread::spawn(move || {
                loop {
                    match stopped.recv_timeout(Duration::from_millis(200)) {
                        Ok(()) | Err(mpsc::RecvTimeoutError::Disconnected) => return false,
                        Err(mpsc::RecvTimeoutError::Timeout) => {}
                    }
                    if fs::OpenOptions::new()
                        .write(true)
                        .custom_flags(libc::O_NONBLOCK)
                        .open(&fifo)
                        .is_ok()
                    {
                        return true;
                    }
                }
            });
            let result = if path == &server.files.port {
                Some(server.read_port())
            } else {
                assert_eq!(super::read_bounded_lossy(path), "");
                None
            };
            let _ = stop_writer.send(());
            assert!(
                !writer.join().expect("join delayed FIFO writer"),
                "startup reader waited for a writer: {}",
                path.display(),
            );
            if let Some(result) = result {
                assert!(matches!(
                    result,
                    Err(PocketIcStartupError::Io { source, .. })
                        if source.kind() == std::io::ErrorKind::InvalidData
                ));
            }
        }
    }

    #[test]
    fn port_file_readiness_preserves_partial_writes_and_rejects_oversized_contents() {
        let (files, _, _) = StartupFiles::create().expect("allocate startup files");
        let server = super::ManagedServer {
            child: None,
            binary: PathBuf::from("unused-server"),
            files,
            started: Instant::now(),
        };
        assert!(matches!(
            server.read_port().unwrap(),
            super::PortFileState::Pending
        ));
        for contents in ["", "34567"] {
            fs::write(&server.files.port, contents).unwrap();
            assert!(matches!(
                server.read_port().unwrap(),
                super::PortFileState::Pending
            ));
        }
        for (contents, expected) in [("1\n", 1), ("65535\n", 65535), (" 34567\r\n", 34567)] {
            fs::write(&server.files.port, contents).unwrap();
            assert!(matches!(
                server.read_port().unwrap(),
                super::PortFileState::Ready(port) if port == expected
            ));
        }
        for contents in ["0\n", "65536\n", "invalid\n", "1\n2\n"] {
            fs::write(&server.files.port, contents).unwrap();
            assert!(matches!(
                server.read_port().unwrap(),
                super::PortFileState::Invalid(_)
            ));
        }
        fs::write(&server.files.port, [0xff, b'\n']).unwrap();
        assert!(matches!(
            server.read_port(),
            Err(PocketIcStartupError::Io { source, .. })
                if source.kind() == std::io::ErrorKind::InvalidData
        ));
        for contents in ["1\n".to_owned() + &" ".repeat(128), "0".repeat(128)] {
            fs::write(&server.files.port, contents).unwrap();
            assert!(
                matches!(
                    server.read_port().unwrap(),
                    super::PortFileState::Invalid(_)
                ),
                "oversized port contents must fail even without a newline",
            );
        }
        // A large backing file must not enlarge the returned diagnostic.
        fs::File::options()
            .write(true)
            .open(&server.files.port)
            .unwrap()
            .set_len(1024 * 1024)
            .unwrap();
        assert!(matches!(
            server.read_port().unwrap(),
            super::PortFileState::Invalid(value) if value.len() < 256
        ));
    }

    #[test]
    fn startup_config_requires_positive_bounds() {
        let error = PocketIcStartupConfig::connect("http://127.0.0.1:1/", Duration::ZERO)
            .validate()
            .expect_err("zero startup timeout must fail");
        assert!(matches!(
            error,
            PocketIcStartupError::InvalidConfiguration { .. }
        ));

        let error = PocketIcStartupConfig::spawn("pocket-ic", Duration::from_secs(1))
            .with_server_hard_ttl(Duration::from_millis(1))
            .validate()
            .expect_err("subsecond server hard TTL must fail");
        assert!(matches!(
            error,
            PocketIcStartupError::InvalidConfiguration { .. }
        ));
    }

    #[test]
    fn managed_server_hard_ttl_is_opt_in() {
        let default = PocketIcStartupConfig::spawn("pocket-ic", Duration::from_secs(1));
        assert_eq!(default.server_hard_ttl(), None);

        let explicit = default.with_server_hard_ttl(Duration::from_secs(17));
        assert_eq!(explicit.server_hard_ttl(), Some(Duration::from_secs(17)));
    }

    #[test]
    fn startup_files_leave_the_server_owned_port_path_absent() {
        let (files, stdout, stderr) = StartupFiles::create().expect("allocate startup files");
        let directory = files.directory.clone();

        assert!(directory.is_dir());
        assert!(!files.port.exists());
        assert!(files.stdout.is_file());
        assert!(files.stderr.is_file());
        #[cfg(unix)]
        {
            let mode = fs::metadata(&directory)
                .expect("inspect private startup directory")
                .permissions()
                .mode();
            assert_eq!(mode & 0o077, 0);
        }

        drop(stdout);
        drop(stderr);
        drop(files);
        assert!(!directory.exists());
    }

    #[cfg(unix)]
    #[test]
    fn managed_startup_reports_an_exited_server_with_bounded_output() {
        let script = TestServerScript::new(
            "exit",
            "#!/bin/sh\nif [ \"$1\" != \"--port-file\" ] || [ -e \"$2\" ]; then exit 97; fi\nprintf 'synthetic server stdout'\nprintf 'synthetic bind failure' >&2\nexit 23\n",
        );

        let result = PocketIcBuilder::new().with_application_subnet().try_build(
            PocketIcStartupConfig::spawn(script.path(), Duration::from_secs(2)),
        );

        let Err(PocketIcStartupError::ServerExited {
            server_binary,
            status,
            stdout,
            stderr,
            ..
        }) = result
        else {
            panic!(
                "an exited managed server must return a structured exit error; got {:?}",
                result.err(),
            );
        };
        assert_eq!(server_binary, script.path());
        assert_eq!(status.code(), Some(23));
        assert_eq!(stdout, "synthetic server stdout");
        assert_eq!(stderr, "synthetic bind failure");
    }

    #[cfg(unix)]
    #[test]
    fn managed_startup_rejects_oversized_port_files_and_cleans_up() {
        let script = TestServerScript::new(
            "oversized-port",
            "#!/bin/sh\nprintf '%s\\n%s\\n' \"$$\" \"$2\"\nprintf '34567\\n%064s' '' > \"$2.pending\"\nmv \"$2.pending\" \"$2\"\nexec sleep 30\n",
        );
        let result = PocketIcStartupConfig::spawn(script.path(), Duration::from_secs(2))
            .start_managed_server();
        let Err(PocketIcStartupError::InvalidServerPort { value, stdout, .. }) = result else {
            panic!("oversized port publication must fail readiness");
        };
        assert!(value.contains("port file exceeds"));
        assert!(value.len() < 256);
        let mut lines = stdout.lines();
        let pid = lines.next().unwrap().parse::<u32>().unwrap();
        let port_path = PathBuf::from(lines.next().unwrap());
        assert!(!port_path.parent().unwrap().exists());
        assert_eq!(process_state(pid), None, "failed server must be reaped");
    }

    #[cfg(unix)]
    #[test]
    fn managed_startup_terminates_a_server_that_never_becomes_ready() {
        let script = TestServerScript::new(
            "timeout",
            "#!/bin/sh\nif [ \"$1\" != \"--port-file\" ] || [ -e \"$2\" ]; then exit 97; fi\nexec sleep 30\n",
        );
        let timeout = Duration::from_millis(100);
        let started = Instant::now();

        let result = PocketIcBuilder::new()
            .with_application_subnet()
            .try_build(PocketIcStartupConfig::spawn(script.path(), timeout));

        assert!(
            started.elapsed() < Duration::from_secs(2),
            "bounded startup should not wait for the sleeping child"
        );
        assert!(matches!(
            result,
            Err(PocketIcStartupError::ReadinessTimeout {
                server_binary,
                timeout: actual_timeout,
                termination_error: None,
                ..
            }) if server_binary == script.path() && actual_timeout == timeout
        ));
    }

    #[cfg(unix)]
    #[test]
    fn managed_server_handle_exposes_process_id_url_output_and_raii_ownership() {
        let script = TestServerScript::new(
            "handle",
            "#!/bin/sh\nif [ \"$1\" != \"--port-file\" ] || [ -e \"$2\" ]; then echo 'unexpected managed server arguments' >&2; exit 97; fi\nprintf 'managed server ready: %s' \"$$\"\nprintf '34567\\n' > \"$2\"\nexec sleep 30\n",
        );

        let server = PocketIcStartupConfig::spawn(script.path(), Duration::from_secs(2))
            .start_managed_server()
            .expect("start caller-owned managed server");

        assert_eq!(server.url(), "http://127.0.0.1:34567/");
        assert_eq!(
            server.output().stdout(),
            format!("managed server ready: {}", server.process_id())
        );
        assert_eq!(server.output().stderr(), "");
        let pid = server.process_id();
        assert!(process_state(pid).is_some_and(|state| state != 'Z'));
        drop(server);
        assert_eq!(process_state(pid), None, "owned server must be reaped");
    }

    #[cfg(unix)]
    #[test]
    fn managed_server_cleans_descendants_on_drop_timeout_exit_and_background_reap() {
        for mode in ["drop", "timeout", "exit", "background"] {
            let publish = if matches!(mode, "drop" | "background") {
                "printf '34567\\n' > \"$2\"\n"
            } else {
                ""
            };
            let finish = if mode == "background" {
                // The caller removes the port only after handing off to the
                // reaper, so slow native hosts cannot miss this ready server.
                "while [ -e \"$2\" ]; do sleep 0.01; done\nexit 23\n"
            } else if mode == "exit" {
                "sleep 0.03\nexit 23\n"
            } else {
                "exec sleep 30\n"
            };
            let script = TestServerScript::new(
                mode,
                &format!("#!/bin/sh\nsleep 30 &\nprintf '%s' \"$!\"\n{publish}{finish}"),
            );
            let result = PocketIcStartupConfig::spawn(script.path(), Duration::from_millis(300))
                .start_managed_server();
            let output = match result {
                Ok(server) => {
                    let output = server.output().stdout().to_owned();
                    if mode == "background" {
                        let port = server.server.files.port.clone();
                        server.server.reap_in_background();
                        fs::remove_file(port).expect("release the background server after handoff");
                    } else {
                        drop(server);
                    }
                    output
                }
                Err(PocketIcStartupError::ReadinessTimeout {
                    stdout,
                    termination_error,
                    ..
                }) => {
                    assert_eq!(mode, "timeout");
                    assert_eq!(termination_error, None);
                    stdout
                }
                Err(PocketIcStartupError::ServerExited { stdout, status, .. }) => {
                    assert_eq!(mode, "exit");
                    assert_eq!(status.code(), Some(23));
                    stdout
                }
                other => panic!(
                    "unexpected {mode} startup result: {}",
                    match other {
                        Err(error) => error.to_string(),
                        Ok(_) => unreachable!(),
                    }
                ),
            };
            let pid = output
                .parse::<u32>()
                .expect("server published its descendant PID");
            let deadline = Instant::now() + Duration::from_secs(2);
            while process_state(pid).is_some_and(|state| state != 'Z') {
                assert!(
                    Instant::now() < deadline,
                    "{mode} left its descendant running"
                );
                std::thread::sleep(Duration::from_millis(10));
            }
        }
    }

    #[cfg(unix)]
    #[test]
    fn managed_server_passes_an_explicit_hard_ttl() {
        let script = TestServerScript::new(
            "hard-ttl",
            "#!/bin/sh\nif [ \"$1\" != \"--hard-ttl\" ] || [ \"$2\" != \"17\" ] || [ \"$3\" != \"--port-file\" ] || [ -e \"$4\" ]; then exit 97; fi\nprintf '34567\\n' > \"$4\"\nexec sleep 30\n",
        );

        let server = PocketIcStartupConfig::spawn(script.path(), Duration::from_secs(2))
            .with_server_hard_ttl(Duration::from_secs(17))
            .start_managed_server()
            .expect("start managed server with an explicit hard TTL");

        assert_eq!(server.url(), "http://127.0.0.1:34567/");
    }

    #[test]
    #[ignore = "requires IC_TESTKIT_POCKET_IC_SERVER=<caller-provided PocketIC server binary>"]
    fn caller_provided_server_publishes_port_constructs_instance_and_cleans_up() {
        let binary = std::env::var_os("IC_TESTKIT_POCKET_IC_SERVER")
            .map(PathBuf::from)
            .expect("set IC_TESTKIT_POCKET_IC_SERVER to the exact server binary");
        let one_shot_sequence = super::STARTUP_FILE_SEQUENCE.load(super::Ordering::Relaxed);
        let one_shot_directory = std::env::temp_dir().join(format!(
            "ic-testkit-pocket-ic-startup-{}-{one_shot_sequence}",
            std::process::id()
        ));
        let one_shot = PocketIcBuilder::new()
            .with_application_subnet()
            .try_build(
                PocketIcStartupConfig::spawn(&binary, Duration::from_secs(30))
                    .with_server_hard_ttl(Duration::from_secs(1)),
            )
            .expect("one-shot managed spawn must construct an instance");
        assert!(one_shot_directory.is_dir());
        drop(one_shot);
        let cleanup_deadline = Instant::now() + Duration::from_secs(3);
        while one_shot_directory.exists() && Instant::now() < cleanup_deadline {
            std::thread::sleep(Duration::from_millis(20));
        }
        assert!(!one_shot_directory.exists());

        let server = PocketIcStartupConfig::spawn(&binary, Duration::from_secs(30))
            .with_server_hard_ttl(Duration::from_secs(60))
            .start_managed_server()
            .expect("caller-provided PocketIC server must publish its port");
        let files = &server.server.files;
        let startup_directory = files.directory.clone();

        assert!(files.port.is_file());
        let pocket_ic = PocketIcBuilder::new()
            .with_application_subnet()
            .try_build(PocketIcStartupConfig::connect(
                server.url(),
                Duration::from_secs(30),
            ))
            .expect("construct instance through caller-provided server");

        drop(pocket_ic);
        drop(server);
        assert!(!startup_directory.exists());
    }

    #[cfg(unix)]
    struct TestServerScript {
        path: PathBuf,
    }

    #[cfg(unix)]
    impl TestServerScript {
        fn new(label: &str, contents: &str) -> Self {
            let path = std::env::temp_dir().join(format!(
                "ic-testkit-pocket-ic-{label}-{}-{}",
                std::process::id(),
                super::STARTUP_FILE_SEQUENCE.fetch_add(1, super::Ordering::Relaxed),
            ));
            write_executable_script(&path, contents);
            Self { path }
        }

        fn path(&self) -> PathBuf {
            self.path.clone()
        }
    }

    #[cfg(unix)]
    impl Drop for TestServerScript {
        fn drop(&mut self) {
            let _ = fs::remove_file(&self.path);
        }
    }
}