running_process/broker/server/

spawn_coordinator.rs

//! Spawn coordination contract for broker-managed backends.
//!
//! This module does not launch child processes yet. It owns the state that
//! Phase 4/5 launch code needs before spawning: per-backend-key budget windows,
//! single-flight protection, retry-after hints for refused Hello replies, and
//! process-wide file locks for backend spawn ownership.

use std::collections::HashMap;
use std::fs::File;
use std::io;
use std::path::{Path, PathBuf};
use std::time::{Duration, Instant};

use super::backend_registry::BackendKey;

/// Default backend spawn attempts allowed per budget window.
pub const DEFAULT_SPAWN_ATTEMPTS_PER_WINDOW: u32 = 3;

/// Default backend spawn budget window.
pub const DEFAULT_SPAWN_BUDGET_WINDOW: Duration = Duration::from_secs(30);

/// Acquire the backend spawn lock at `path`.
///
/// The returned guard owns an exclusive OS file lock until it is dropped. The
/// lock file is intentionally left in place on drop; ownership is attached to
/// the open file handle, not to path existence.
///
/// On Unix and Windows this helper verifies file identity after taking the
/// lock. If another coordinator deletes, renames, or recreates the lock file
/// between open and lock acquisition, the helper refuses the stale handle with
/// [`SpawnLockError::DeletedOrRecreated`]. On platforms where file identity is
/// not available, callers must keep lock files in a trusted broker-owned
/// directory and treat path deletion/recreation detection as best-effort.
pub fn acquire_spawn_lock(path: impl AsRef<Path>) -> Result<SpawnLockGuard, SpawnLockError> {
    acquire_spawn_lock_with_hook(path.as_ref(), |_, _| {})
}

fn acquire_spawn_lock_with_hook<F>(
    path: &Path,
    mut before_lock: F,
) -> Result<SpawnLockGuard, SpawnLockError>
where
    F: FnMut(&Path, &File),
{
    let path_buf = path.to_path_buf();
    let file =
        crate::platform::fs::open_lock_file(path).map_err(|source| SpawnLockError::Open {
            path: path_buf.clone(),
            source,
        })?;

    before_lock(path, &file);

    crate::platform::fs::try_lock_exclusive(&file).map_err(|source| {
        if crate::platform::fs::is_lock_conflict(&source) {
            SpawnLockError::AlreadyLocked {
                path: path_buf.clone(),
            }
        } else {
            SpawnLockError::Lock {
                path: path_buf.clone(),
                source,
            }
        }
    })?;

    let opened_identity = crate::platform::fs::file_identity(&file)
        .map_err(|source| lock_identity_error(&path_buf, &file, source))?;
    let current_identity = match crate::platform::fs::path_identity(path) {
        Ok(identity) => identity,
        Err(source) if source.kind() == io::ErrorKind::NotFound => {
            let _ = crate::platform::fs::unlock(&file);
            return Err(SpawnLockError::DeletedOrRecreated {
                path: path_buf,
                opened_identity,
                current_identity: None,
            });
        }
        Err(source) => return Err(lock_identity_error(&path_buf, &file, source)),
    };

    if opened_identity != current_identity {
        let _ = crate::platform::fs::unlock(&file);
        return Err(SpawnLockError::DeletedOrRecreated {
            path: path_buf,
            opened_identity,
            current_identity,
        });
    }

    Ok(SpawnLockGuard {
        file,
        path: path_buf,
        identity: opened_identity,
    })
}

fn lock_identity_error(path: &Path, file: &File, source: io::Error) -> SpawnLockError {
    let _ = crate::platform::fs::unlock(file);
    SpawnLockError::Identity {
        path: path.to_path_buf(),
        source,
    }
}

/// RAII guard for an acquired backend spawn lock.
#[must_use = "dropping the guard releases the backend spawn lock immediately"]
#[derive(Debug)]
pub struct SpawnLockGuard {
    file: File,
    path: PathBuf,
    identity: Option<SpawnLockFileIdentity>,
}

impl SpawnLockGuard {
    /// Lock file path that was acquired.
    pub fn path(&self) -> &Path {
        &self.path
    }

    /// Platform file identity captured for the lock file, when available.
    pub fn file_identity(&self) -> Option<SpawnLockFileIdentity> {
        self.identity
    }
}

impl Drop for SpawnLockGuard {
    fn drop(&mut self) {
        let _ = crate::platform::fs::unlock(&self.file);
    }
}

/// Stable identity for an opened lock file on platforms that expose it.
///
/// Kept under this name for callers that already match on it; the mechanic is
/// the facade's, because "is this still the same file" is a question only the
/// host can answer.
pub use crate::platform::fs::FileIdentity as SpawnLockFileIdentity;

/// Errors returned while acquiring a backend spawn lock.
#[derive(Debug, thiserror::Error)]
pub enum SpawnLockError {
    /// The lock path could not be opened or created.
    #[error("failed to open backend spawn lock file {path}: {source}")]
    Open {
        /// Lock path.
        path: PathBuf,
        /// Underlying I/O error.
        #[source]
        source: io::Error,
    },
    /// Another broker worker already owns the lock.
    #[error("backend spawn lock file {path} is already locked")]
    AlreadyLocked {
        /// Lock path.
        path: PathBuf,
    },
    /// The platform lock operation failed for a reason other than contention.
    #[error("failed to lock backend spawn lock file {path}: {source}")]
    Lock {
        /// Lock path.
        path: PathBuf,
        /// Underlying I/O error.
        #[source]
        source: io::Error,
    },
    /// The lock path no longer names the file that was locked.
    #[error("backend spawn lock file {path} was deleted or recreated during acquisition")]
    DeletedOrRecreated {
        /// Lock path.
        path: PathBuf,
        /// Identity of the opened file handle.
        opened_identity: Option<SpawnLockFileIdentity>,
        /// Identity currently reachable through the lock path.
        current_identity: Option<SpawnLockFileIdentity>,
    },
    /// File identity could not be read.
    #[error("failed to verify backend spawn lock file identity for {path}: {source}")]
    Identity {
        /// Lock path.
        path: PathBuf,
        /// Underlying I/O error.
        #[source]
        source: io::Error,
    },
}

/// Spawn-budget tuning.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct SpawnBudgetConfig {
    /// Maximum spawn attempts in one window.
    pub max_attempts: u32,
    /// Window duration.
    pub window: Duration,
}

impl SpawnBudgetConfig {
    /// Build a config, clamping zero values to safe non-zero defaults.
    pub fn new(max_attempts: u32, window: Duration) -> Self {
        Self {
            max_attempts: max_attempts.max(1),
            window: if window.is_zero() {
                Duration::from_millis(1)
            } else {
                window
            },
        }
    }
}

impl Default for SpawnBudgetConfig {
    fn default() -> Self {
        Self {
            max_attempts: DEFAULT_SPAWN_ATTEMPTS_PER_WINDOW,
            window: DEFAULT_SPAWN_BUDGET_WINDOW,
        }
    }
}

/// Coordinates bounded spawn attempts for backend keys.
#[derive(Debug)]
pub struct SpawnCoordinator {
    config: SpawnBudgetConfig,
    states: HashMap<BackendKey, SpawnBudgetState>,
}

impl SpawnCoordinator {
    /// Create an empty coordinator with default budget settings.
    pub fn new() -> Self {
        Self::with_config(SpawnBudgetConfig::default())
    }

    /// Create an empty coordinator with explicit budget settings.
    pub fn with_config(config: SpawnBudgetConfig) -> Self {
        Self {
            config,
            states: HashMap::new(),
        }
    }

    /// Begin one spawn attempt for `key`.
    ///
    /// The returned permit is a contract token for the caller that will perform
    /// the actual child-process launch in later slices. Call [`Self::finish`]
    /// when that launch path succeeds or fails.
    pub fn try_begin(
        &mut self,
        key: BackendKey,
        now: Instant,
    ) -> Result<SpawnPermit, SpawnBeginError> {
        let state = self
            .states
            .entry(key.clone())
            .or_insert_with(|| SpawnBudgetState::new(now));
        state.refresh(now, self.config.window);

        if state.in_flight {
            return Err(SpawnBeginError::AlreadyInProgress);
        }

        if state.attempts_used >= self.config.max_attempts {
            let is_storm_trip = !state.storm_signaled;
            state.storm_signaled = true;
            return Err(SpawnBeginError::BudgetExhausted {
                retry_after: retry_after(state.window_started_at, now, self.config.window),
                remaining: 0,
                is_storm_trip,
            });
        }

        state.attempts_used += 1;
        state.in_flight = true;
        Ok(SpawnPermit {
            key,
            attempt_number: state.attempts_used,
            remaining_after_begin: self.config.max_attempts - state.attempts_used,
        })
    }

    /// Finish an in-flight spawn attempt.
    pub fn finish(&mut self, key: &BackendKey, outcome: SpawnOutcome, now: Instant) {
        let Some(state) = self.states.get_mut(key) else {
            return;
        };
        state.refresh(now, self.config.window);
        state.in_flight = false;
        if outcome == SpawnOutcome::Success {
            state.window_started_at = now;
            state.attempts_used = 0;
            state.storm_signaled = false;
        }
    }

    /// Return the current budget snapshot for one backend key.
    pub fn snapshot(&mut self, key: BackendKey, now: Instant) -> SpawnBudgetSnapshot {
        let state = self
            .states
            .entry(key.clone())
            .or_insert_with(|| SpawnBudgetState::new(now));
        state.refresh(now, self.config.window);
        snapshot_for(key, state, self.config, now)
    }
}

impl Default for SpawnCoordinator {
    fn default() -> Self {
        Self::new()
    }
}

/// Token returned for a spawn attempt that may proceed.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct SpawnPermit {
    /// Backend key this permit covers.
    pub key: BackendKey,
    /// 1-based attempt number inside the current window.
    pub attempt_number: u32,
    /// Budget remaining after this attempt starts.
    pub remaining_after_begin: u32,
}

/// Result of a spawn attempt.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum SpawnOutcome {
    /// The backend process was launched and verified.
    Success,
    /// The backend process failed to launch or verify.
    Failed,
}

/// Errors returned when a spawn attempt cannot begin.
#[derive(Clone, Debug, PartialEq, Eq, thiserror::Error)]
pub enum SpawnBeginError {
    /// Another worker is already launching this backend key.
    #[error("backend spawn already in progress")]
    AlreadyInProgress,
    /// The per-key spawn budget is exhausted.
    #[error("backend spawn budget exhausted; retry after {retry_after:?}")]
    BudgetExhausted {
        /// Time until the budget window resets.
        retry_after: Duration,
        /// Remaining attempts, always zero for this variant.
        remaining: u32,
        /// True on the FIRST `try_begin` call to observe this key's budget
        /// as exhausted within the current window; false on every
        /// subsequent call until the window resets or a spawn succeeds.
        /// A caller that reacts to the spawn-storm signal (e.g. rotating
        /// the broker token) should gate on this to react exactly once
        /// per storm, not once per refused request during it.
        is_storm_trip: bool,
    },
}

/// Current budget state for metrics/admin snapshots.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct SpawnBudgetSnapshot {
    /// Backend key this snapshot describes.
    pub key: BackendKey,
    /// Attempts used in the active window.
    pub attempts_used: u32,
    /// Attempts still available in the active window.
    pub remaining: u32,
    /// Whether a spawn is currently in flight.
    pub in_flight: bool,
    /// Retry-after hint when no attempts remain.
    pub retry_after: Option<Duration>,
}

#[derive(Clone, Debug)]
struct SpawnBudgetState {
    window_started_at: Instant,
    attempts_used: u32,
    in_flight: bool,
    /// Whether this key's budget exhaustion has already been reported as a
    /// storm trip in the current window (zackees/soldr#2364: "driving M
    /// failed daemon spawns inside the window trips the guard exactly
    /// once"). Every `try_begin` call after the first exhaustion keeps
    /// returning `BudgetExhausted`, so a caller reacting to the FIRST one
    /// (e.g. rotating the broker token) needs this to avoid reacting again
    /// on every subsequent call in the same window.
    storm_signaled: bool,
}

impl SpawnBudgetState {
    fn new(now: Instant) -> Self {
        Self {
            window_started_at: now,
            attempts_used: 0,
            in_flight: false,
            storm_signaled: false,
        }
    }

    fn refresh(&mut self, now: Instant, window: Duration) {
        if elapsed_since(self.window_started_at, now) >= window {
            self.window_started_at = now;
            self.attempts_used = 0;
            self.in_flight = false;
            self.storm_signaled = false;
        }
    }
}

fn snapshot_for(
    key: BackendKey,
    state: &SpawnBudgetState,
    config: SpawnBudgetConfig,
    now: Instant,
) -> SpawnBudgetSnapshot {
    let remaining = config.max_attempts.saturating_sub(state.attempts_used);
    SpawnBudgetSnapshot {
        key,
        attempts_used: state.attempts_used,
        remaining,
        in_flight: state.in_flight,
        retry_after: (remaining == 0)
            .then(|| retry_after(state.window_started_at, now, config.window)),
    }
}

fn retry_after(window_started_at: Instant, now: Instant, window: Duration) -> Duration {
    window.saturating_sub(elapsed_since(window_started_at, now))
}

fn elapsed_since(started_at: Instant, now: Instant) -> Duration {
    now.checked_duration_since(started_at)
        .unwrap_or(Duration::ZERO)
}

#[cfg(test)]
mod tests {
    use std::fs;

    use super::*;

    #[test]
    #[cfg(any(unix, windows))]
    fn acquire_spawn_lock_detects_lock_file_replacement_between_open_and_lock() {
        let tmp = tempfile::tempdir().unwrap();
        let lock_path = tmp.path().join("backend.spawn.lock");
        let replaced_path = tmp.path().join("backend.spawn.lock.replaced");

        let err = acquire_spawn_lock_with_hook(&lock_path, |path, _file| {
            fs::rename(path, &replaced_path).unwrap();
            fs::write(path, b"replacement lock file").unwrap();
        })
        .unwrap_err();

        let SpawnLockError::DeletedOrRecreated {
            path,
            opened_identity: Some(opened_identity),
            current_identity: Some(current_identity),
        } = err
        else {
            panic!("expected deleted/recreated error, got {err:?}");
        };

        assert_eq!(path, lock_path);
        assert_ne!(opened_identity, current_identity);

        let _guard = acquire_spawn_lock(&lock_path).unwrap();
    }
}