henad-explore 0.3.0

//! Checks that a resumed sweep keeps its finished runs, reruns the rest and ends with the files of a sweep in one go.

use std::fs;
use std::path::Path;
use std::time::Duration;

use henad_compute::entry::register_grid_model;
use henad_core::explore::design::DesignKind;
use henad_core::explore::factor::{FactorSpec, LevelSpec};
use henad_core::explore::plan::Shard;
use henad_core::explore::spec::{ActionSpec, BlockSpec, SweepSpec};

use crate::exec::Concurrency;
use crate::output::manifest::{BuildRole, ManifestStatus, RecordedBuild};
use crate::output::resume::ResumeError;
use crate::output::{MANIFEST_FILE, OutputError, RUNS_FILE, SERIES_FILE, SUMMARY_FILE};
use crate::sweep::{ExploreError, SweepOptions, SweepWarning};
use crate::tests::broken::DividesByParam;
use crate::tests::support::{
    OutputTables, Recorder, ScratchDir, SecondResume, dry_run, entry, manifest, other_engine, provenance,
    rewrite_manifest, sweep, sweep_options, sweep_with,
};

fn values(raw: &[&str]) -> LevelSpec {
    LevelSpec::Values(raw.iter().map(|&text| text.to_owned()).collect())
}

/// Returns 3 configs of SIR on a 16 by 16 grid with `replicates` replicates, and an action whose tick varies.
fn sir_spec(replicates: u64) -> SweepSpec {
    let mut spec = SweepSpec::new("sir");
    spec.fixed = vec![
        ("grid_width".to_owned(), "16".to_owned()),
        ("grid_height".to_owned(), "16".to_owned()),
    ];
    spec.run.steps = 30;
    spec.run.replicates = replicates;
    spec.measure.stats_every = 3;
    spec.measure.series_every = 6;
    spec.seeds.root = 8;
    spec.actions = vec![ActionSpec::new("seed_outbreak", 12)];
    spec.blocks = vec![BlockSpec {
        design: DesignKind::Zip,
        factors: vec![
            FactorSpec::param("infection_rate", values(&["0.2", "0.4", "0.6"])),
            FactorSpec::action("seed_outbreak", values(&["6", "12", "18"])),
        ],
        design_seed: None,
    }];
    spec
}

/// Runs a resume of `spec` into `output_dir` with `options`, and returns the ids of the runs it committed.
fn resume(entry_id: &str, spec: &SweepSpec, output_dir: &Path, options: SweepOptions) -> Vec<u64> {
    let model = entry(entry_id, None);
    let mut progress = Recorder::default();
    let options = SweepOptions {
        resume: true,
        ..options
    };
    sweep_with(&model, None, spec, output_dir, &options, &mut progress).expect("the sweep resumes");
    progress.committed
}

/// Returns the bytes of each of the four files of the output directory `dir`, by file name.
fn file_bytes(dir: &Path) -> Vec<(&'static str, Vec<u8>)> {
    [RUNS_FILE, SERIES_FILE, SUMMARY_FILE, MANIFEST_FILE]
        .into_iter()
        .map(|file| (file, fs::read(dir.join(file)).expect("the file is written")))
        .collect()
}

/// Returns the text of the first `line_count` lines of `text`.
fn first_lines(text: &str, line_count: usize) -> &str {
    let end = text
        .match_indices('\n')
        .nth(line_count - 1)
        .map_or(text.len(), |(index, _)| index + 1);
    &text[..end]
}

#[test]
fn a_resumed_sweep_skips_finished_runs_and_matches_a_fresh_one() {
    let sir = entry("sir", None);
    let spec = sir_spec(2);
    let scratch = ScratchDir::new("resume");
    let fresh_dir = scratch.path().join("fresh");
    let resumed_dir = scratch.path().join("resumed");
    sweep(&sir, None, &spec, &fresh_dir, Concurrency::Auto);
    sweep(&sir, None, &spec, &resumed_dir, Concurrency::Auto);

    // A process that ended while writing run 2 leaves its row cut short, and the series rows of later runs.
    let runs = fs::read_to_string(resumed_dir.join(RUNS_FILE)).expect("runs.csv is written");
    let kept = first_lines(&runs, 3);
    let partial = &first_lines(&runs, 4)[kept.len()..][..20];
    fs::write(resumed_dir.join(RUNS_FILE), format!("{kept}{partial}")).expect("runs.csv is cut");
    let mut series = fs::read_to_string(resumed_dir.join(SERIES_FILE)).expect("series.csv is written");
    series.push_str("5,1");
    fs::write(resumed_dir.join(SERIES_FILE), series).expect("series.csv gains a partial line");

    let committed = resume("sir", &spec, &resumed_dir, sweep_options(false));
    assert_eq!(committed, [2, 3, 4, 5], "runs 0 and 1 were kept");
    assert_eq!(OutputTables::read(&resumed_dir), OutputTables::read(&fresh_dir));
    let recorded = manifest(&resumed_dir);
    assert_eq!(recorded.status, ManifestStatus::Complete);
    let sessions: Vec<(u64, u64)> = recorded
        .sessions
        .iter()
        .map(|session| (session.skipped, session.ran))
        .collect();
    assert_eq!(sessions, [(0, 6), (2, 4)]);
    assert_eq!(recorded.results.map(|results| results.rows), Some(6));

    assert!(
        resume("sir", &spec, &resumed_dir, sweep_options(false)).is_empty(),
        "a complete directory has nothing left to run"
    );
    assert_eq!(OutputTables::read(&resumed_dir), OutputTables::read(&fresh_dir));
}

#[test]
fn a_second_resume_is_refused_while_the_first_holds_its_scan() {
    let sir = entry("sir", None);
    let spec = sir_spec(2);
    let scratch = ScratchDir::new("resume-locked");
    let fresh_dir = scratch.path().join("fresh");
    let resumed_dir = scratch.path().join("resumed");
    sweep(&sir, None, &spec, &fresh_dir, Concurrency::Auto);
    sweep(&sir, None, &spec, &resumed_dir, Concurrency::Auto);
    let runs = fs::read_to_string(resumed_dir.join(RUNS_FILE)).expect("runs.csv is written");
    fs::write(resumed_dir.join(RUNS_FILE), first_lines(&runs, 3)).expect("runs.csv is cut");

    let mut second = SecondResume::new(&sir, &spec, &resumed_dir);
    let report =
        sweep_with(&sir, None, &spec, &resumed_dir, &sweep_options(true), &mut second).expect("the first resume runs");
    let refused = second.result.expect("the first resume reports its outline");
    assert!(
        matches!(refused, Err(ExploreError::Output(OutputError::Locked { .. }))),
        "{refused:?}"
    );
    assert_eq!(report.outline.skipped, 2);
    assert_eq!(
        OutputTables::read(&resumed_dir),
        OutputTables::read(&fresh_dir),
        "no run is written twice"
    );
}

#[test]
fn a_resume_credits_a_session_its_process_left_unfinished() {
    let spec = sir_spec(2);
    let scratch = ScratchDir::new("resume-unfinished");
    sweep(&entry("sir", None), None, &spec, scratch.path(), Concurrency::Auto);

    // A process killed after its second run leaves two rows and the manifest it wrote at the start.
    let runs = fs::read_to_string(scratch.path().join(RUNS_FILE)).expect("runs.csv is written");
    fs::write(scratch.path().join(RUNS_FILE), first_lines(&runs, 3)).expect("runs.csv is cut");
    rewrite_manifest(scratch.path(), |recorded| {
        recorded.status = ManifestStatus::Running;
        recorded.results = None;
        recorded.sessions[0].ran = 0;
    });

    assert_eq!(resume("sir", &spec, scratch.path(), sweep_options(false)), [2, 3, 4, 5]);
    let sessions: Vec<(u64, u64)> = manifest(scratch.path())
        .sessions
        .iter()
        .map(|session| (session.skipped, session.ran))
        .collect();
    assert_eq!(sessions, [(0, 2), (2, 4)]);
}

#[test]
fn a_resume_names_a_changed_model() {
    let sir = entry("sir", None);
    let spec = sir_spec(2);
    let scratch = ScratchDir::new("resume-model");
    sweep(&sir, None, &spec, scratch.path(), Concurrency::Auto);

    // Another version of the model changes the schema hash, and with it the plan hash.
    rewrite_manifest(scratch.path(), |recorded| {
        recorded.model.schema_hash = "0000000000000000".to_owned();
        recorded.plan.plan_hash = "0000000000000000".to_owned();
    });
    let error = sweep_with(
        &sir,
        None,
        &spec,
        scratch.path(),
        &sweep_options(true),
        &mut Recorder::default(),
    )
    .expect_err("another model");
    assert!(
        matches!(error, ExploreError::Resume(ResumeError::SchemaChanged { .. })),
        "{error:?}"
    );
}

#[test]
fn a_dry_run_counts_the_runs_a_resume_skips_and_changes_nothing() {
    let sir = entry("sir", None);
    let spec = sir_spec(2);
    let scratch = ScratchDir::new("resume-dry-run");
    sweep(&sir, None, &spec, scratch.path(), Concurrency::Auto);
    let runs = fs::read_to_string(scratch.path().join(RUNS_FILE)).expect("runs.csv is written");
    let cut = first_lines(&runs, 4).to_owned();
    fs::write(scratch.path().join(RUNS_FILE), &cut).expect("runs.csv is cut");
    let series = fs::read(scratch.path().join(SERIES_FILE)).expect("series.csv is written");

    let mut options = sweep_options(true);
    options.provenance = provenance().with_engine(other_engine());
    let mut progress = Recorder::default();
    let report = dry_run(&sir, &spec, Some(scratch.path()), &options, &mut progress).expect("the dry run plans");
    assert_eq!((report.outline.skipped, report.outline.pending), (3, 3));
    assert_eq!(
        progress.warnings,
        [SweepWarning::BuildChanged {
            role: BuildRole::Engine,
            recorded: Box::new(RecordedBuild::engine()),
            current: Box::new(other_engine()),
            between_shards: false,
        }]
    );
    assert_eq!(
        fs::read_to_string(scratch.path().join(RUNS_FILE)).expect("runs.csv is kept"),
        cut
    );
    assert_eq!(
        fs::read(scratch.path().join(SERIES_FILE)).expect("series.csv is kept"),
        series,
        "a dry run leaves the orphan rows for the resume"
    );
}

#[test]
fn resume_refuses_a_changed_spec() {
    let sir = entry("sir", None);
    let spec = sir_spec(2);
    let scratch = ScratchDir::new("resume-changed");
    sweep(&sir, None, &spec, scratch.path(), Concurrency::Auto);
    let written = file_bytes(scratch.path());

    let mut longer = spec.clone();
    longer.run.steps = 31;
    let mut other_seed = spec.clone();
    other_seed.seeds.root = 9;
    let mut moved = spec.clone();
    moved.blocks[0].factors[1] = FactorSpec::action("seed_outbreak", values(&["6", "12", "19"]));
    for changed in [longer, other_seed, moved] {
        let error = sweep_with(
            &sir,
            None,
            &changed,
            scratch.path(),
            &sweep_options(true),
            &mut Recorder::default(),
        )
        .expect_err("another plan");
        assert!(
            matches!(error, ExploreError::Resume(ResumeError::PlanChanged { .. })),
            "{error:?}"
        );
    }
    let sharded = SweepOptions {
        shard: Shard::new(1, 2).expect("a valid shard"),
        ..sweep_options(true)
    };
    let error =
        sweep_with(&sir, None, &spec, scratch.path(), &sharded, &mut Recorder::default()).expect_err("another shard");
    assert!(
        matches!(error, ExploreError::Resume(ResumeError::ShardChanged { .. })),
        "{error:?}"
    );
    for ((file, before), (_, after)) in written.iter().zip(file_bytes(scratch.path())) {
        assert!(*before == after, "a refused resume leaves {file} as it was");
    }

    let mut timed = spec.clone();
    timed.run.timeout = Some(Duration::from_secs(600));
    assert!(
        resume("sir", &timed, scratch.path(), sweep_options(false)).is_empty(),
        "the timeout is no part of the plan"
    );
}

#[test]
fn retry_failed_reruns_only_failures() {
    let model = register_grid_model::<DividesByParam>();
    let mut spec = SweepSpec::new("divides_by_param");
    spec.fixed = vec![
        ("grid_width".to_owned(), "8".to_owned()),
        ("grid_height".to_owned(), "8".to_owned()),
    ];
    spec.run.steps = 10;
    spec.run.replicates = 2;
    spec.measure.stats_every = 2;
    spec.measure.series_every = 2;
    spec.blocks = vec![BlockSpec {
        design: DesignKind::Factorial,
        factors: vec![FactorSpec::param("divisor", values(&["1", "0", "2"]))],
        design_seed: None,
    }];
    let scratch = ScratchDir::new("retry-failed");
    sweep(&model, None, &spec, scratch.path(), Concurrency::Auto);
    let first = OutputTables::read(scratch.path());
    assert_eq!(
        first.run_column("status"),
        ["ok", "ok", "panicked", "panicked", "ok", "ok"]
    );

    let resumed = |retry_failed: bool| {
        let options = SweepOptions {
            retry_failed,
            ..sweep_options(true)
        };
        let mut progress = Recorder::default();
        sweep_with(&model, None, &spec, scratch.path(), &options, &mut progress).expect("the sweep resumes");
        progress.committed
    };
    assert!(resumed(false).is_empty(), "a failed run counts as finished");
    assert_eq!(resumed(true), [2, 3]);
    assert_eq!(
        OutputTables::read(scratch.path()),
        first,
        "the retried runs fail as they did"
    );
    let sessions: Vec<(u64, u64)> = manifest(scratch.path())
        .sessions
        .iter()
        .map(|session| (session.skipped, session.ran))
        .collect();
    assert_eq!(sessions, [(0, 6), (6, 0), (4, 2)]);
}

#[test]
fn raising_replicates_on_resume_runs_only_the_new_ones() {
    let sir = entry("sir", None);
    let scratch = ScratchDir::new("more-replicates");
    let resumed_dir = scratch.path().join("resumed");
    sweep(&sir, None, &sir_spec(2), &resumed_dir, Concurrency::Auto);
    let committed = resume("sir", &sir_spec(3), &resumed_dir, sweep_options(false));
    assert_eq!(committed, [2, 5, 8], "replicate 2 of each config");

    let fresh_dir = scratch.path().join("fresh");
    sweep(&sir, None, &sir_spec(3), &fresh_dir, Concurrency::Auto);
    let resumed = OutputTables::read(&resumed_dir);
    assert_eq!(resumed, OutputTables::read(&fresh_dir));
    assert_eq!(resumed.run_column("rep"), ["0", "1", "2", "0", "1", "2", "0", "1", "2"]);
    let recorded = manifest(&resumed_dir);
    assert_eq!((recorded.plan.replicates, recorded.plan.runs), (3, 9));

    // Only run 0, replicate 0 of config 0, stays, so every kept row fits the lower count.
    let runs = fs::read_to_string(resumed_dir.join(RUNS_FILE)).expect("runs.csv is written");
    fs::write(resumed_dir.join(RUNS_FILE), first_lines(&runs, 2)).expect("runs.csv is cut");
    let error = sweep_with(
        &sir,
        None,
        &sir_spec(1),
        &resumed_dir,
        &sweep_options(true),
        &mut Recorder::default(),
    )
    .expect_err("fewer replicates");
    assert!(
        matches!(
            error,
            ExploreError::Resume(ResumeError::ReplicatesLowered {
                recorded: 3,
                current: 1
            })
        ),
        "{error:?}"
    );
}

#[test]
fn a_run_past_its_timeout_is_recorded_and_resume_retries_it() {
    let sir = entry("sir", None);
    let scratch = ScratchDir::new("timeout");
    let mut timed = sir_spec(2);
    timed.run.timeout = Some(Duration::ZERO);
    let resumed_dir = scratch.path().join("resumed");
    let report = sweep(&sir, None, &timed, &resumed_dir, Concurrency::Auto);
    assert_eq!((report.counts.rows, report.counts.failed), (6, 6));
    let timed_out = OutputTables::read(&resumed_dir);
    assert!(
        timed_out
            .run_column("status")
            .iter()
            .all(|&status| status == "timed_out")
    );
    assert!(
        timed_out
            .run_column("stop_reason")
            .iter()
            .all(|&reason| reason == "timeout")
    );
    for ticks in timed_out.run_column("ticks") {
        assert!(ticks.parse::<u64>().is_ok_and(|ticks| ticks < 30), "{ticks}");
    }
    assert!(timed_out.run_column("note")[0].starts_with("timed out after 0 s"));

    let committed = resume("sir", &sir_spec(2), &resumed_dir, sweep_options(false));
    assert_eq!(committed, [0, 1, 2, 3, 4, 5], "a timed-out run is never finished");
    let fresh_dir = scratch.path().join("fresh");
    sweep(&sir, None, &sir_spec(2), &fresh_dir, Concurrency::Auto);
    assert_eq!(OutputTables::read(&resumed_dir), OutputTables::read(&fresh_dir));
}