effectfence 0.3.0

//! The fence must outlive the process that runs it.
//!
//! An in-memory ledger is exactly as strong as the proxy's uptime: restart it
//! and every recorded outcome and every in-flight lease vanish, so the next
//! identical call runs the effect again -- the double-execution this crate
//! exists to prevent, reintroduced by a deploy. These tests open a fence on a
//! file, drop it, open it again, and check what the second process knows.

use effectfence::fence::{
    Admission, EffectFence, EffectRequest, FenceConfig, FenceError, VectorClock, abort_effect,
    commit_effect_cert, prepare_effect_fence,
};
use std::path::PathBuf;
use std::time::Duration;

fn ledger_path(tag: &str) -> PathBuf {
    let nanos = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap()
        .as_nanos();
    let p = std::env::temp_dir().join(format!(
        "effectfence-{tag}-{}-{nanos}.jsonl",
        std::process::id()
    ));
    let _ = std::fs::remove_file(&p);
    p
}

fn req(intent: &str, domain: &str) -> EffectRequest {
    EffectRequest {
        intent: intent.to_string(),
        parent: None,
        domain: domain.to_string(),
        tool: "charge_card".to_string(),
        args: serde_json::json!({"amount_cents": 4900}),
        read_set: vec![],
        agent: "agent-a".to_string(),
        known_clock: VectorClock::new(),
    }
}

#[test]
fn a_committed_result_is_replayed_after_restart() {
    let path = ledger_path("done");
    let first_hash = {
        let fence = EffectFence::open(&path, FenceConfig::default()).expect("open ledger");
        let prepared = match prepare_effect_fence(&fence, req("charge:order-1", "order:1")) {
            Ok(Admission::Fresh(p)) => p,
            other => panic!("expected Fresh, got {other:?}"),
        };
        commit_effect_cert(&fence, prepared, serde_json::json!({"charged": 4900}))
            .expect("commit")
            .hash
    }; // process one ends here

    let fence = EffectFence::open(&path, FenceConfig::default()).expect("reopen ledger");
    match prepare_effect_fence(&fence, req("charge:order-1", "order:1")) {
        Ok(Admission::Replay(cert)) => assert_eq!(cert.hash, first_hash),
        other => panic!("a restarted fence must replay the recorded result, got {other:?}"),
    }
    let _ = std::fs::remove_file(&path);
}

#[test]
fn an_effect_in_flight_at_restart_is_fenced_not_rerun() {
    let path = ledger_path("inflight");
    {
        let fence = EffectFence::open(&path, FenceConfig::default()).expect("open ledger");
        let _running = prepare_effect_fence(&fence, req("charge:order-2", "order:2"))
            .expect("first admission");
        // The process dies here with the provider call outstanding.
    }

    let fence = EffectFence::open(&path, FenceConfig::default()).expect("reopen ledger");
    match prepare_effect_fence(&fence, req("charge:order-2", "order:2")) {
        Err(FenceError::IntentFailed { reason, .. }) => assert!(
            reason.contains("restart"),
            "the reason should say the holder's process restarted, got: {reason}"
        ),
        other => panic!(
            "an intent that was mid-flight when the process died has an UNKNOWN outcome \
             and must be fenced, got {other:?}"
        ),
    }
    let _ = std::fs::remove_file(&path);
}

#[test]
fn a_failed_outcome_stays_fenced_after_restart() {
    let path = ledger_path("failed");
    {
        let fence = EffectFence::open(&path, FenceConfig::default()).expect("open ledger");
        let prepared = match prepare_effect_fence(&fence, req("charge:order-3", "order:3")) {
            Ok(Admission::Fresh(p)) => p,
            other => panic!("expected Fresh, got {other:?}"),
        };
        abort_effect(
            &fence,
            prepared,
            "provider timed out after the request was sent",
        );
    }

    let fence = EffectFence::open(&path, FenceConfig::default()).expect("reopen ledger");
    assert!(
        matches!(
            prepare_effect_fence(&fence, req("charge:order-3", "order:3")),
            Err(FenceError::IntentFailed { .. })
        ),
        "a fenced intent must still be fenced after a restart"
    );
    let _ = std::fs::remove_file(&path);
}

#[test]
fn an_operator_clear_survives_restart() {
    let path = ledger_path("cleared");
    {
        let fence = EffectFence::open(&path, FenceConfig::default()).expect("open ledger");
        let prepared = match prepare_effect_fence(&fence, req("charge:order-4", "order:4")) {
            Ok(Admission::Fresh(p)) => p,
            other => panic!("expected Fresh, got {other:?}"),
        };
        abort_effect(&fence, prepared, "unknown");
        assert!(
            fence.clear_intent("charge:order-4"),
            "operator reconciled and cleared"
        );
    }

    let fence = EffectFence::open(&path, FenceConfig::default()).expect("reopen ledger");
    assert!(
        matches!(
            prepare_effect_fence(&fence, req("charge:order-4", "order:4")),
            Ok(Admission::Fresh(_))
        ),
        "an intent the operator cleared before the restart must be runnable after it"
    );
    let _ = std::fs::remove_file(&path);
}

#[test]
fn domain_sequence_continues_after_restart() {
    let path = ledger_path("domain");
    {
        let fence = EffectFence::open(&path, FenceConfig::default()).expect("open ledger");
        let prepared = match prepare_effect_fence(&fence, req("charge:order-5a", "order:5")) {
            Ok(Admission::Fresh(p)) => p,
            other => panic!("expected Fresh, got {other:?}"),
        };
        assert_eq!(prepared.seq, 1);
        commit_effect_cert(&fence, prepared, serde_json::json!({})).unwrap();
    }

    let fence = EffectFence::open(&path, FenceConfig::default()).expect("reopen ledger");
    assert_eq!(
        fence.current("order:5"),
        1,
        "the domain's causal position is state too"
    );
    match prepare_effect_fence(&fence, req("charge:order-5b", "order:5")) {
        Ok(Admission::Fresh(p)) => assert_eq!(p.seq, 2, "a restart must not restart the order"),
        other => panic!("expected Fresh, got {other:?}"),
    }
    let _ = std::fs::remove_file(&path);
}

#[test]
fn an_expired_result_is_not_replayed_after_restart() {
    let path = ledger_path("ttl");
    let config = FenceConfig {
        result_ttl: Duration::from_millis(20),
        ..FenceConfig::default()
    };
    {
        let fence = EffectFence::open(&path, config).expect("open ledger");
        let prepared = match prepare_effect_fence(&fence, req("charge:order-6", "order:6")) {
            Ok(Admission::Fresh(p)) => p,
            other => panic!("expected Fresh, got {other:?}"),
        };
        commit_effect_cert(&fence, prepared, serde_json::json!({})).unwrap();
    }
    std::thread::sleep(Duration::from_millis(60));

    let fence = EffectFence::open(&path, config).expect("reopen ledger");
    assert!(
        matches!(
            prepare_effect_fence(&fence, req("charge:order-6", "order:6")),
            Ok(Admission::Fresh(_))
        ),
        "the replay horizon is wall-clock time, not process uptime"
    );
    let _ = std::fs::remove_file(&path);
}

#[test]
fn a_truncated_last_line_does_not_prevent_opening() {
    let path = ledger_path("torn");
    {
        let fence = EffectFence::open(&path, FenceConfig::default()).expect("open ledger");
        let prepared = match prepare_effect_fence(&fence, req("charge:order-7", "order:7")) {
            Ok(Admission::Fresh(p)) => p,
            other => panic!("expected Fresh, got {other:?}"),
        };
        commit_effect_cert(&fence, prepared, serde_json::json!({})).unwrap();
    }
    // A crash mid-write leaves a torn trailing line.
    {
        use std::io::Write;
        let mut f = std::fs::OpenOptions::new()
            .append(true)
            .open(&path)
            .unwrap();
        f.write_all(b"{\"op\":\"done\",\"intent\":\"charge:ord")
            .unwrap();
    }

    let fence = EffectFence::open(&path, FenceConfig::default()).expect("a torn tail is tolerated");
    assert!(
        matches!(
            prepare_effect_fence(&fence, req("charge:order-7", "order:7")),
            Ok(Admission::Replay(_))
        ),
        "everything before the torn line is still known"
    );
    let _ = std::fs::remove_file(&path);
}