oxo-flow-cli 0.29.0

//! Env-knob parsing and small path helpers: spawn-watchdog + disk-pressure tunables, end-of-run prune decision, path absolutize/tmp_dir/normalize (#948 P2-5).

use super::*;

// ── Spawn watchdog (issue #685) ─────────────────────────────────────────────
// A rule can end up marked Running in the scheduler while its task never
// reaches a child spawn (lost wakeup in the wait/dispatch path). The engine
// then wedges: nothing completes, nothing fails, the process idles in a
// futex wait. The watchdog compares the scheduler's running set against the
// executor's live child pids on a fixed tick and reports stragglers. Since
// #847, rules queued in the resource pool are reported as benign waits
// (WARN) — only unqueued rules keep the #685 ERROR. Since #872 the watchdog
// re-arms on every tick once a rule is past the threshold, and after
// `WATCHDOG_REQUEUE_FIRES` consecutive fires on the same rule it requeues
// the rule instead of waiting forever.

/// How often the watchdog compares running rules against live pids.
pub(crate) const SPAWN_WATCHDOG_TICK_SECS: u64 = 30;
/// Default age at which a Running rule with no live child process is
/// reported. Generous: legitimate env setup can take tens of minutes.
pub(crate) const DEFAULT_SPAWN_WATCHDOG_SECS: u64 = 600;
pub(crate) const SPAWN_WATCHDOG_ENV: &str = "OXO_FLOW_SPAWN_WATCHDOG_SECS";
/// Consecutive watchdog fires on one unqueued rule after which the rule is
/// requeued instead of being left to wedge its lane forever (issue #872).
/// The first fire lands at the threshold (600 s default); subsequent fires
/// land on every 30 s tick, so 6 fires ≈ 600 s + 5×30 s ≈ 13 minutes of
/// zero dispatch progress at the defaults.
pub(crate) const WATCHDOG_REQUEUE_FIRES: u32 = 6;
/// Environment override for `WATCHDOG_REQUEUE_FIRES`. Same convention as
/// the other watchdog knobs: unparseable or `0` falls back to the default —
/// a typo can never disable the requeue (it would wedge the lane instead).
pub(crate) const WATCHDOG_REQUEUE_FIRES_ENV: &str = "OXO_FLOW_SPAWN_WATCHDOG_REQUEUE_FIRES";

pub(crate) fn watchdog_requeue_fires() -> u32 {
    parse_watchdog_requeue_fires(std::env::var(WATCHDOG_REQUEUE_FIRES_ENV).ok().as_deref())
}

pub(crate) fn parse_watchdog_requeue_fires(raw: Option<&str>) -> u32 {
    raw.and_then(|v| v.parse::<u32>().ok())
        .filter(|&n| n > 0)
        .unwrap_or(WATCHDOG_REQUEUE_FIRES)
}

/// How often the #843 disk-pressure monitor re-measures free space while the
/// run is parked under Hold (nothing in flight) or between completion batches.
pub(crate) const DISK_PRESSURE_TICK_SECS: u64 = 60;
/// Consecutive parked Hold rounds (nothing in flight, nothing reclaimable)
/// after which the run aborts: the disk is not recovering on its own.
pub(crate) const DISK_HOLD_ABORT_ROUNDS: u32 = 2;
/// Environment overrides for the two constants above (review: shared-HPC
/// scratch may want a longer grace than 2×60s). Same convention as
/// `OXO_FLOW_SPAWN_WATCHDOG_SECS`: unparseable or `0` falls back to the
/// default — a typo can never disable the abort (it would hang instead).
pub(crate) const DISK_TICK_ENV: &str = "OXO_FLOW_DISK_TICK_SECS";
pub(crate) const DISK_HOLD_ROUNDS_ENV: &str = "OXO_FLOW_DISK_HOLD_ROUNDS";

pub(crate) fn disk_tick_secs() -> u64 {
    parse_disk_tick_secs(std::env::var(DISK_TICK_ENV).ok().as_deref())
}

pub(crate) fn parse_disk_tick_secs(raw: Option<&str>) -> u64 {
    raw.and_then(|v| v.parse::<u64>().ok())
        .filter(|&s| s > 0)
        .unwrap_or(DISK_PRESSURE_TICK_SECS)
}

pub(crate) fn disk_hold_abort_rounds() -> u32 {
    parse_disk_hold_abort_rounds(std::env::var(DISK_HOLD_ROUNDS_ENV).ok().as_deref())
}

pub(crate) fn parse_disk_hold_abort_rounds(raw: Option<&str>) -> u32 {
    raw.and_then(|v| v.parse::<u32>().ok())
        .filter(|&s| s > 0)
        .unwrap_or(DISK_HOLD_ABORT_ROUNDS)
}

/// Parse the watchdog threshold from the environment. Extracted as a pure
/// function of the env so tests exercise it without process-global mutation.
pub(crate) fn spawn_watchdog_threshold() -> std::time::Duration {
    parse_spawn_watchdog_threshold(std::env::var(SPAWN_WATCHDOG_ENV).ok().as_deref())
}

/// Pure derivation: `None`/unparseable/`0` fall back to the default — `0`
/// matches the `OXO_ENV_LOCK_TIMEOUT_SECS` convention rather than meaning
/// "disable", so a typo can never silently turn the watchdog off.
pub(crate) fn parse_spawn_watchdog_threshold(raw: Option<&str>) -> std::time::Duration {
    raw.and_then(|v| v.parse::<u64>().ok())
        .filter(|&s| s > 0)
        .map(std::time::Duration::from_secs)
        .unwrap_or(std::time::Duration::from_secs(DEFAULT_SPAWN_WATCHDOG_SECS))
}

/// Pure decision for the end-of-run temporary sweep (#893): should the
/// fully-successful run prune temporary outputs?
///
/// - No `[engine]` thresholds configured → sweep unconditionally (the
///   documented historical behavior; disk pressure mid-run is the only
///   reclaim path).
/// - Thresholds configured and free space measured at/above
///   `reclaim_free_disk` → skip the sweep (disk pressure alone triggers
///   re-materialization, not every run).
/// - Thresholds configured but the measurement is unavailable (non-Unix) →
///   keep outputs, matching `DiskPressureMonitor`'s fail-open stance.
pub(crate) fn endofrun_prune_decision(
    free_mb: Option<u64>,
    thresholds: Option<(u64, u64)>,
) -> EndOfRunPruneDecision {
    let Some((_, reclaim_mb)) = thresholds else {
        return EndOfRunPruneDecision::Sweep;
    };
    match free_mb {
        Some(free) if free >= reclaim_mb => EndOfRunPruneDecision::Skip {
            free_mb: free,
            reclaim_mb,
        },
        Some(_) => EndOfRunPruneDecision::Sweep,
        None => EndOfRunPruneDecision::KeepOnUnknown,
    }
}

/// Outcome of [`endofrun_prune_decision`] — see that function for the
/// semantics of each arm.
#[derive(Debug)]
pub(crate) enum EndOfRunPruneDecision {
    /// Run the post-run temporary sweep (historical behavior).
    Sweep,
    /// Free space is at/above the reclaim floor — keep temporary outputs.
    Skip { free_mb: u64, reclaim_mb: u64 },
    /// Free space could not be measured — keep temporary outputs (fail-open).
    KeepOnUnknown,
}
/// Resolve a possibly-relative path against the current directory without
/// resolving symlinks. Checkpoint records use this so `resume` works from
/// any invocation directory (issue #68).
pub(crate) fn absolutize(path: &Path) -> Result<PathBuf> {
    if path.is_absolute() {
        Ok(path.to_path_buf())
    } else {
        Ok(std::env::current_dir()?.join(path))
    }
}

/// Resolve the run's global scratch directory (`[engine] tmp_dir`, issue
/// #907). `OXO_FLOW_TMPDIR` overrides the workflow value — the same
/// env-beats-config pattern as `OXO_FLOW_OUTPUT_TAIL_BYTES` — so operators
/// can redirect temp traffic per invocation without touching the file.
/// A relative workflow value anchors to the workflow directory (the same
/// root as file-backed env specs, issue #427); the configured directory is
/// created on demand so the first rule doesn't race on mkdir. `None`
/// (unset everywhere) keeps the ambient environment untouched.
pub(crate) fn tmp_dir_resolved(
    configured: Option<String>,
    workflow_dir: &Path,
) -> Result<Option<PathBuf>> {
    if let Ok(raw) = std::env::var("OXO_FLOW_TMPDIR") {
        let raw = raw.trim().to_string();
        if !raw.is_empty() {
            let resolved = absolutize(Path::new(&raw))?;
            // Same fail-fast mkdir as the config path below — the docs
            // promise "created on demand at run start" for both routes.
            std::fs::create_dir_all(&resolved)
                .with_context(|| format!("creating OXO_FLOW_TMPDIR dir {}", resolved.display()))?;
            return Ok(Some(resolved));
        }
        tracing::warn!("OXO_FLOW_TMPDIR is empty, ignoring it in favor of [engine] tmp_dir");
    }
    let Some(raw) = configured else {
        return Ok(None);
    };
    let trimmed = raw.trim();
    if trimmed.is_empty() {
        tracing::warn!("[engine] tmp_dir is empty, ignoring it");
        return Ok(None);
    }
    let raw_path = Path::new(trimmed);
    // Anchor on the RAW value — absolutizing first would join the process
    // cwd and silently turn every relative config into a cwd-relative path.
    let anchored = if raw_path.is_absolute() {
        raw_path.to_path_buf()
    } else {
        workflow_dir.join(raw_path)
    };
    std::fs::create_dir_all(&anchored)
        .with_context(|| format!("creating [engine] tmp_dir {}", anchored.display()))?;
    Ok(Some(anchored))
}

/// Lexically collapse `.`/`..` segments from an absolutized path
/// (`/wf/./data` → `/wf/data`) without touching the filesystem. The
/// workflow-dir join in `parent_dir` leaves a `.` segment for bare
/// filenames; anchored {input}/{log} strings inherit it (issue #427).
pub(crate) fn normalize_dots(path: PathBuf) -> PathBuf {
    let mut out = PathBuf::with_capacity(path.as_os_str().len());
    for comp in path.components() {
        match comp {
            std::path::Component::CurDir => {}
            std::path::Component::ParentDir => {
                // Only reachable for the leading segments of a relative
                // input — absolutize already stripped interior `..` against
                // a real base. Keep it verbatim rather than guessing.
                out.push(comp);
            }
            other => out.push(other),
        }
    }
    if out.as_os_str().is_empty() {
        PathBuf::from(".")
    } else {
        out
    }
}

#[cfg(test)]
mod tests {
    //! `[engine] tmp_dir` resolution (issue #907): env override wins,
    //! relative values anchor to the workflow dir, unset stays unset.

    use super::EndOfRunPruneDecision as Prune;
    use super::*;

    #[test]
    fn unset_everywhere_is_none() {
        // OXO_FLOW_TMPDIR is unset in CI and normal dev shells; if a developer
        // exports it, skip rather than fail — the test asserts the CONFIGURED
        // path, not the operator's machine.
        if std::env::var("OXO_FLOW_TMPDIR").is_ok() {
            return;
        }
        let resolved = tmp_dir_resolved(None, std::path::Path::new("/wf")).unwrap();
        assert!(resolved.is_none(), "no config + no env = None");
    }

    #[test]
    fn relative_config_anchors_to_workflow_dir() {
        let dir = std::env::temp_dir().join(format!("oxo-tmpdircfg-{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&dir);
        std::fs::create_dir_all(&dir).unwrap();
        let workflow_dir = dir.join("wf");
        std::fs::create_dir_all(&workflow_dir).unwrap();

        let resolved = tmp_dir_resolved(Some("scratch".to_string()), &workflow_dir).unwrap();
        assert_eq!(
            resolved.as_deref(),
            Some(workflow_dir.join("scratch").as_path()),
            "relative tmp_dir must anchor to the workflow directory"
        );
        // Created on demand so the first rule doesn't race on mkdir.
        assert!(workflow_dir.join("scratch").is_dir());

        let _ = std::fs::remove_dir_all(&dir);
    }

    #[test]
    fn empty_config_value_is_ignored() {
        if std::env::var("OXO_FLOW_TMPDIR").is_ok() {
            return;
        }
        let resolved =
            tmp_dir_resolved(Some("   ".to_string()), std::path::Path::new("/wf")).unwrap();
        assert!(resolved.is_none(), "empty/whitespace tmp_dir = None");
    }

    #[test]
    fn parse_disk_env_knobs_fall_back_to_defaults() {
        // Unset, empty, garbage, zero — a typo can never disable the
        // abort (it would hang instead), so every invalid form falls
        // back to the shipped default.
        assert_eq!(parse_disk_tick_secs(None), 60);
        assert_eq!(parse_disk_tick_secs(Some("")), 60);
        assert_eq!(parse_disk_tick_secs(Some("abc")), 60);
        assert_eq!(parse_disk_tick_secs(Some("0")), 60);
        assert_eq!(parse_disk_tick_secs(Some("-5")), 60);
        assert_eq!(parse_disk_hold_abort_rounds(None), 2);
        assert_eq!(parse_disk_hold_abort_rounds(Some("xyz")), 2);
        assert_eq!(parse_disk_hold_abort_rounds(Some("0")), 2);
    }

    #[test]
    fn parse_disk_env_knobs_honor_valid_values() {
        assert_eq!(parse_disk_tick_secs(Some("1")), 1);
        assert_eq!(parse_disk_tick_secs(Some("300")), 300);
        assert_eq!(parse_disk_hold_abort_rounds(Some("1")), 1);
        assert_eq!(parse_disk_hold_abort_rounds(Some("10")), 10);
    }

    #[test]
    fn endofrun_prune_sweeps_without_engine_thresholds() {
        // No `[engine]` section → the historical unconditional sweep, on
        // every free-space outcome (the measurement is simply irrelevant).
        assert!(matches!(
            endofrun_prune_decision(Some(0), None),
            Prune::Sweep
        ));
        assert!(matches!(
            endofrun_prune_decision(Some(39_000_000), None),
            Prune::Sweep
        ));
        assert!(matches!(endofrun_prune_decision(None, None), Prune::Sweep));
    }

    #[test]
    fn endofrun_prune_skips_above_reclaim_floor() {
        // #893: with thresholds configured, comfortable free space keeps
        // temporary outputs — the same floor the mid-run reclaim uses.
        let thresholds = Some((5_000_000, 5_000_000)); // 5T reclaim
        match endofrun_prune_decision(Some(39_000_000), thresholds) {
            Prune::Skip {
                free_mb,
                reclaim_mb,
            } => {
                assert_eq!(free_mb, 39_000_000);
                assert_eq!(reclaim_mb, 5_000_000);
            }
            other => panic!("expected Skip, got {other:?}"),
        }
        // Exactly at the floor counts as above it (mid-run reclaim only
        // fires BELOW reclaim_free_disk — mirror that boundary).
        assert!(matches!(
            endofrun_prune_decision(Some(5_000_000), thresholds),
            Prune::Skip { .. }
        ));
    }

    #[test]
    fn endofrun_prune_sweeps_below_reclaim_floor() {
        // Under pressure → sweep (delete temporaries to free the space).
        assert!(matches!(
            endofrun_prune_decision(Some(4_999_999), Some((5_000_000, 5_000_000))),
            Prune::Sweep
        ));
    }

    #[test]
    fn endofrun_prune_keeps_outputs_when_measurement_unknown() {
        // Fail-open matches DiskPressureMonitor: a non-Unix platform under
        // configured thresholds must NOT sweep (an unknown disk state is
        // not a license to delete).
        assert!(matches!(
            endofrun_prune_decision(None, Some((5_000_000, 5_000_000))),
            Prune::KeepOnUnknown
        ));
    }

    #[test]
    fn endofrun_prune_debug_representations_are_distinct() {
        // Guards the enum from collapsing arms (e.g. Skip accidentally
        // constructed for the KeepOnUnknown case) in a debug-print context.
        let skip = format!(
            "{:?}",
            Prune::Skip {
                free_mb: 1,
                reclaim_mb: 1
            }
        );
        let unknown = format!("{:?}", Prune::KeepOnUnknown);
        let sweep = format!("{:?}", Prune::Sweep);
        assert_ne!(skip, unknown);
        assert_ne!(skip, sweep);
        assert_ne!(unknown, sweep);
    }

    #[test]
    fn spawn_watchdog_threshold_parsing() {
        // issue #685: unset, garbage, and `0` all fall back to the default
        // — `0` matches the OXO_ENV_LOCK_TIMEOUT_SECS convention rather
        // than meaning "disable", so a typo can never silently turn the
        // watchdog off. The derivation is pure: no env mutation needed.
        assert_eq!(
            parse_spawn_watchdog_threshold(None),
            std::time::Duration::from_secs(600)
        );
        assert_eq!(
            parse_spawn_watchdog_threshold(Some("abc")),
            std::time::Duration::from_secs(600)
        );
        assert_eq!(
            parse_spawn_watchdog_threshold(Some("0")),
            std::time::Duration::from_secs(600)
        );
        assert_eq!(
            parse_spawn_watchdog_threshold(Some("120")),
            std::time::Duration::from_secs(120)
        );
    }

    #[test]
    fn watchdog_requeue_fires_parsing() {
        // issue #872: same convention as the threshold — unset, garbage, and
        // `0` all fall back to the default so a typo can never disable the
        // requeue (it would wedge the lane instead). Pure derivation.
        assert_eq!(parse_watchdog_requeue_fires(None), 6);
        assert_eq!(parse_watchdog_requeue_fires(Some("abc")), 6);
        assert_eq!(parse_watchdog_requeue_fires(Some("0")), 6);
        assert_eq!(parse_watchdog_requeue_fires(Some("3")), 3);
    }
}