haematite 0.6.1

//! Sol r3 fold pins: the anchor's no-follow/causal-creation discipline and
//! the validated `shard_count` (zero and absurd counts refuse typed, never
//! silently skip M1 or abort at the allocator).

use std::error::Error;
use std::fs;

use super::error::VacuumError;
use super::test_support::{build_store, expect_refusal, snapshot_tree, stats};

/// A pre-existing `writer.lock` SYMLINK refuses — dangling AND resolving —
/// and creates NOTHING outside the data dir. `O_CREAT` through a dangling
/// link would mint the target inode; the no-follow guard runs first.
#[cfg(unix)]
#[test]
fn symlinked_anchor_refuses_and_mints_nothing() -> Result<(), Box<dyn Error>> {
    let temp = tempfile::tempdir()?;
    build_store(temp.path(), 1, &[b"data"], 1)?;
    let lock_path = temp.path().join("writer.lock");
    fs::remove_file(&lock_path)?;
    let external = tempfile::tempdir()?;

    // Dangling: the classic mint-an-inode-elsewhere shape.
    let target = external.path().join("minted-elsewhere");
    std::os::unix::fs::symlink(&target, &lock_path)?;
    let error = expect_refusal(stats(temp.path()))?;
    let text = error.to_string();
    assert!(
        matches!(
            &error,
            VacuumError::LockIo {
                anchor_created: false,
                ..
            }
        ),
        "got {error:?}"
    );
    assert!(text.contains("regular file"), "names the shape: {text}");
    assert!(text.contains("re-run"), "names the remedy: {text}");
    assert!(
        !target.exists(),
        "NOTHING may be created through the dangling link"
    );

    // Resolving: an existing external file must not be opened or locked.
    fs::remove_file(&lock_path)?;
    let resolving = external.path().join("real-file");
    fs::write(&resolving, b"external")?;
    std::os::unix::fs::symlink(&resolving, &lock_path)?;
    let before = fs::metadata(&resolving)?.modified()?;
    let error = expect_refusal(stats(temp.path()))?;
    assert!(
        matches!(
            &error,
            VacuumError::LockIo {
                anchor_created: false,
                ..
            }
        ),
        "got {error:?}"
    );
    assert_eq!(fs::read(&resolving)?, b"external");
    assert_eq!(fs::metadata(&resolving)?.modified()?, before);
    Ok(())
}

/// Anchor creation is CAUSAL: created exactly when this run's atomic
/// `create_new` succeeded — true on the anchor-less run, false when the
/// anchor pre-exists (including the one the previous run just made).
#[test]
fn anchor_creation_flag_is_causal() -> Result<(), Box<dyn Error>> {
    let temp = tempfile::tempdir()?;
    build_store(temp.path(), 1, &[b"data"], 1)?;

    let report = stats(temp.path())?;
    assert!(!report.lock_anchor_created, "create() made the anchor");

    fs::remove_file(temp.path().join("writer.lock"))?;
    let report = stats(temp.path())?;
    assert!(report.lock_anchor_created, "this run created it");

    let report = stats(temp.path())?;
    assert!(
        !report.lock_anchor_created,
        "the previous run's anchor pre-exists now"
    );
    Ok(())
}

fn write_raw_config(data_dir: &std::path::Path, shard_count: &str) -> Result<(), Box<dyn Error>> {
    fs::write(
        data_dir.join("config.json"),
        format!(
            "{{\"format_version\":1,\"data_dir\":{:?},\"shard_count\":{shard_count},\
             \"distributed\":null}}",
            data_dir.display().to_string()
        ),
    )?;
    Ok(())
}

/// `shard_count: 0` parses but is INVALID by the engine's own rules — and
/// would make stats silently skip the exhaustive M1 source. Typed refusal,
/// zero filesystem changes (no anchor on the anchor-less dir).
#[test]
fn zero_shard_count_refuses_without_writing() -> Result<(), Box<dyn Error>> {
    let temp = tempfile::tempdir()?;
    write_raw_config(temp.path(), "0")?;

    let before = snapshot_tree(temp.path())?;
    let error = expect_refusal(stats(temp.path()))?;
    match &error {
        VacuumError::ConfigUnreadable { reason, .. } => {
            assert!(reason.contains("shard_count"), "{reason}");
        }
        other => return Err(format!("expected ConfigUnreadable, got {other:?}").into()),
    }
    assert_eq!(before, snapshot_tree(temp.path())?, "zero changes");
    Ok(())
}

/// Sol r4: the create-then-lose-the-race interleaving — this run's
/// `create_new` makes the anchor, a competitor grabs the OS lock before our
/// `try_lock`, and the Locked refusal still DISCLOSES the write (the causal
/// fact survives the contention arm).
#[test]
fn losing_the_race_after_creating_the_anchor_is_disclosed() -> Result<(), Box<dyn Error>> {
    let temp = tempfile::tempdir()?;
    build_store(temp.path(), 1, &[b"data"], 1)?;
    fs::remove_file(temp.path().join("writer.lock"))?;

    crate::db::lock::contend_next_try_lock();
    let error = expect_refusal(stats(temp.path()));
    crate::db::lock::release_contenders();
    let error = error?;
    match &error {
        VacuumError::Locked { anchor_created, .. } => {
            assert!(anchor_created, "the causal creation fact must survive");
        }
        other => return Err(format!("expected Locked, got {other:?}").into()),
    }
    let text = error.to_string();
    assert!(
        text.contains("CREATED the empty advisory writer.lock"),
        "the Locked refusal must disclose the write: {text}"
    );
    Ok(())
}

/// Sol r4: stats validates ONLY what it consumes. Damaged distributed
/// configuration — a null topology with a zero sync interval — must NOT make
/// the inventory unavailable: `shard_count` is trustworthy, so stats proceeds.
#[test]
fn irrelevant_config_damage_does_not_block_stats() -> Result<(), Box<dyn Error>> {
    let temp = tempfile::tempdir()?;
    build_store(temp.path(), 1, &[b"data"], 2)?;
    // Rewrite config.json with the same shard_count but a broken distributed
    // section (refused by Database::open's own validation and irrelevant to
    // inventory).
    fs::write(
        temp.path().join("config.json"),
        format!(
            "{{\"format_version\":1,\"data_dir\":{:?},\"shard_count\":1,\
             \"distributed\":{{\"local_node\":\"node-a\",\
             \"nodes\":[\"node-a\"],\"topology\":null,\"sync_interval\":0}}}}",
            temp.path().display().to_string()
        ),
    )?;

    let report = stats(temp.path())?;
    assert!(report.totals.enumerated_nodes > 0, "the inventory ran");
    Ok(())
}