vantage-vista 2.6.0

//! A Rhai `atomic {}` block lands every write it makes, or none.

use std::sync::{Arc, Mutex, PoisonError};
use std::time::Duration;

use vantage_core::{Result, error};
use vantage_rhai::rhai::{Dynamic, EvalAltResult, Map, Scope};
use vantage_rhai::{Host, Limits};

use super::transactions::{
    FAILED_STATEMENT, PROBE_DEADLINE, Runner, TxFixture, TxTables, absent, ensure, present, run,
    writable,
};
use crate::rhai::{
    ATOMIC_ARM_TIMEOUT, ATOMIC_BLOCK_TIMEOUT, AtomicScope, AtomicVocab, DataVocab, ResolverCell,
    Stop, Terminals, Writes,
};
use crate::{Vista, VistaResolver};

const NESTED: &str = "atomic can't be nested";
/// Ends a hung block inside its probe's deadline, so it fails the probe
/// rather than the whole test.
const BLOCK_LIMIT: Duration = Duration::from_secs(1);
const ARM_BUDGET: Duration = Duration::from_millis(200);

/// Runs every `atomic` probe on fresh tables, one after another, under the
/// same budget and deadlines as [`check_transactions`](super::check_transactions),
/// then panics once, listing the probes that broke the contract. Scripts see
/// the tables as `a` and `b`, and the second datasource's
/// ([`fresh_other`](TxFixture::fresh_other)) as `c` and `d`, and run on a
/// blocking thread, the way a host runs them; reads "outside" go through the
/// unwrapped Vistas.
pub async fn check_atomic_rhai(fx: &impl TxFixture) {
    let mut runner = Runner::new();
    if !fx.supports_transactions() {
        if writable(fx).await {
            run(
                &mut runner,
                "direct_writes_land_and_are_reported",
                PROBE_DEADLINE,
                direct_writes_land_and_are_reported(fx),
            )
            .await;
        } else {
            run(
                &mut runner,
                "direct_reads_pass_through",
                PROBE_DEADLINE,
                direct_reads_pass_through(fx),
            )
            .await;
        }
        return runner.finish();
    }
    run(
        &mut runner,
        "atomic_block_commits",
        PROBE_DEADLINE,
        atomic_block_commits(fx),
    )
    .await;
    run(
        &mut runner,
        "a_throw_rolls_back_and_runs_the_arm",
        PROBE_DEADLINE,
        a_throw_rolls_back_and_runs_the_arm(fx),
    )
    .await;
    run(
        &mut runner,
        "a_throwing_arm_rethrows_the_first_error",
        PROBE_DEADLINE,
        a_throwing_arm_rethrows_the_first_error(fx),
    )
    .await;
    run(
        &mut runner,
        "return_inside_atomic_commits",
        PROBE_DEADLINE,
        return_inside_atomic_commits(fx),
    )
    .await;
    run(
        &mut runner,
        "a_handle_from_before_the_block_joins_it",
        PROBE_DEADLINE,
        a_handle_from_before_the_block_joins_it(fx),
    )
    .await;
    run(
        &mut runner,
        "a_handle_outlives_its_block",
        PROBE_DEADLINE,
        a_handle_outlives_its_block(fx),
    )
    .await;
    run(
        &mut runner,
        "a_failed_statement_fails_the_block",
        PROBE_DEADLINE,
        a_failed_statement_fails_the_block(fx),
    )
    .await;
    run(
        &mut runner,
        "nested_atomic_is_refused",
        PROBE_DEADLINE,
        nested_atomic_is_refused(fx),
    )
    .await;
    run(
        &mut runner,
        "two_databases_commit_together",
        PROBE_DEADLINE,
        two_databases_commit_together(fx),
    )
    .await;
    run(
        &mut runner,
        "a_throw_rolls_both_back",
        PROBE_DEADLINE,
        a_throw_rolls_both_back(fx),
    )
    .await;
    run(
        &mut runner,
        "committed_is_empty_when_nothing_landed",
        PROBE_DEADLINE,
        committed_is_empty_when_nothing_landed(fx),
    )
    .await;
    runner.finish();
}

async fn atomic_block_commits(fx: &impl TxFixture) -> Result<()> {
    let t = fx.fresh().await;
    let ran = eval(
        &t,
        r#"atomic {
            table("a").insert(#{ id: "x", n: 1 });
            table("b").insert(#{ id: "y", a_id: "x", n: 1 });
        }"#,
    )
    .await?;
    if let Err(e) = ran.out {
        return Err(error!(format!("the block throws `{e}`")));
    }
    present(&t.a, "x", "after the block").await?;
    present(&t.b, "y", "after the block").await
}

async fn a_throw_rolls_back_and_runs_the_arm(fx: &impl TxFixture) -> Result<()> {
    let t = fx.fresh().await;
    let ran = eval(
        &t,
        r#"let seen = ();
        atomic {
            table("a").insert(#{ id: "x", n: 1 });
            throw "stop";
        } rollback (e) {
            seen = e;
        }"#,
    )
    .await?;
    threw(&ran, "stop")?;
    ensure(
        ran.seen.as_deref() == Some("stop"),
        format!("the arm binds {:?}, not \"stop\"", ran.seen),
    )?;
    absent(&t.a, "x", "after the throw").await
}

async fn a_throwing_arm_rethrows_the_first_error(fx: &impl TxFixture) -> Result<()> {
    let t = fx.fresh().await;
    let ran = eval(
        &t,
        r#"atomic {
            table("a").insert(#{ id: "x", n: 1 });
            throw "stop";
        } rollback (e) {
            throw "arm";
        }"#,
    )
    .await?;
    threw(&ran, "stop")?;
    absent(&t.a, "x", "after the arm threw").await
}

async fn return_inside_atomic_commits(fx: &impl TxFixture) -> Result<()> {
    let t = fx.fresh().await;
    let ran = eval(
        &t,
        r#"fn f() {
            atomic {
                table("a").insert(#{ id: "x", n: 1 });
                return 7;
            }
        }
        f()"#,
    )
    .await?;
    match &ran.out {
        Ok(v) if v.as_int() == Ok(7) => {}
        Ok(v) => return Err(error!(format!("f() returns {v}, not 7"))),
        Err(e) => return Err(error!(format!("f() throws `{e}`, not returns 7"))),
    }
    present(&t.a, "x", "after the return").await
}

async fn a_handle_from_before_the_block_joins_it(fx: &impl TxFixture) -> Result<()> {
    let t = fx.fresh().await;
    let ran = eval(
        &t,
        r#"let t = table("a");
        atomic {
            t.insert(#{ id: "x", n: 1 });
            throw "stop";
        }"#,
    )
    .await?;
    threw(&ran, "stop")?;
    absent(&t.a, "x", "after the throw").await
}

/// The handle is made inside the block; read after it, it sees the commit.
async fn a_handle_outlives_its_block(fx: &impl TxFixture) -> Result<()> {
    let t = fx.fresh().await;
    let ran = eval(
        &t,
        r#"let h;
        atomic {
            h = table("a");
            h.insert(#{ id: "x", n: 1 });
        }
        h.get("x")"#,
    )
    .await?;
    let row = ran
        .out
        .map_err(|e| error!(format!("the read after the block throws `{e}`")))?;
    let n = row
        .clone()
        .try_cast::<Map>()
        .and_then(|row| row.get("n").and_then(|n| n.as_int().ok()));
    ensure(
        n == Some(1),
        format!("h.get(\"x\") after the block returns {row}, not a row with n = 1"),
    )?;
    present(&t.a, "x", "after the block").await
}

/// The body catches the failed insert and ends Ok, so only the commit can
/// fail the block.
async fn a_failed_statement_fails_the_block(fx: &impl TxFixture) -> Result<()> {
    let t = fx.fresh().await;
    let ran = eval(
        &t,
        r#"atomic {
            table("a").insert(#{ id: "x", n: 1 });
            try { table("a").insert(#{ id: "z", n: -1 }) } catch {}
        }"#,
    )
    .await?;
    let failed = matches!(&ran.out, Err(e) if e.contains(FAILED_STATEMENT));
    ensure(
        failed,
        format!("the block ends {:?}, not `{FAILED_STATEMENT}`", ran.out),
    )?;
    absent(&t.a, "x", "after the failed block").await
}

async fn nested_atomic_is_refused(fx: &impl TxFixture) -> Result<()> {
    let t = fx.fresh().await;
    let ran = eval(
        &t,
        r#"atomic {
            table("a").insert(#{ id: "x", n: 1 });
            atomic { 1 }
        }"#,
    )
    .await?;
    threw(&ran, NESTED)?;
    absent(&t.a, "x", "after the nested block").await
}

/// The block throws after writing directly: the write stays, and the block
/// reports its table once.
async fn direct_writes_land_and_are_reported(fx: &impl TxFixture) -> Result<()> {
    let t = fx.fresh().await;
    let ran = eval(
        &t,
        r#"atomic {
            table("a").insert(#{ id: "x", n: 1 });
            throw "stop";
        }"#,
    )
    .await?;
    threw(&ran, "stop")?;
    present(&t.a, "x", "after the throw, written directly").await?;
    let wanted = [vec![t.a.name().to_string()]];
    ensure(
        ran.unprotected == wanted,
        format!(
            "the block reports {:?} as unprotected, not {wanted:?}",
            ran.unprotected
        ),
    )
}

/// A read-only datasource without transactions: the block reads as outside,
/// and reports no table.
async fn direct_reads_pass_through(fx: &impl TxFixture) -> Result<()> {
    let t = fx.fresh().await;
    let outside = t.a.count().await?;
    let ran = eval(
        &t,
        r#"let seen = ();
        atomic {
            seen = table("a").count().to_string();
        }"#,
    )
    .await?;
    if let Err(e) = ran.out {
        return Err(error!(format!("the block throws `{e}`")));
    }
    ensure(
        ran.seen == Some(outside.to_string()),
        format!(
            "the block counts {:?} rows of {}, {outside} outside",
            ran.seen,
            t.a.name()
        ),
    )?;
    ensure(
        ran.unprotected.concat().is_empty(),
        format!(
            "a block that only reads reports {:?} as unprotected",
            ran.unprotected
        ),
    )
}

async fn two_databases_commit_together(fx: &impl TxFixture) -> Result<()> {
    let Some(o) = fx.fresh_other().await else {
        return Ok(());
    };
    let t = fx.fresh().await;
    let ran = eval_on(
        &t,
        Some(&o),
        r#"atomic {
            table("a").insert(#{ id: "x", n: 1 });
            table("c").insert(#{ id: "x", n: 1 });
        }"#,
    )
    .await?;
    if let Err(e) = ran.out {
        return Err(error!(format!("the block throws `{e}`")));
    }
    present(&t.a, "x", "after the block").await?;
    present(&o.a, "x", "after the block").await
}

async fn a_throw_rolls_both_back(fx: &impl TxFixture) -> Result<()> {
    let Some(o) = fx.fresh_other().await else {
        return Ok(());
    };
    let t = fx.fresh().await;
    let ran = eval_on(
        &t,
        Some(&o),
        r#"atomic {
            table("a").insert(#{ id: "x", n: 1 });
            table("c").insert(#{ id: "x", n: 1 });
            throw "stop";
        }"#,
    )
    .await?;
    threw(&ran, "stop")?;
    absent(&t.a, "x", "after the throw").await?;
    absent(&o.a, "x", "after the throw").await
}

/// A throw before the commit: no datasource committed, so the arm's
/// `committed` is an empty array.
async fn committed_is_empty_when_nothing_landed(fx: &impl TxFixture) -> Result<()> {
    let Some(o) = fx.fresh_other().await else {
        return Ok(());
    };
    let t = fx.fresh().await;
    let ran = eval_on(
        &t,
        Some(&o),
        r#"let seen = ();
        atomic {
            table("a").insert(#{ id: "x", n: 1 });
            table("c").insert(#{ id: "x", n: 1 });
            throw "stop";
        } rollback (e, committed) {
            seen = `${type_of(committed)} of ${committed.len()}`;
        }"#,
    )
    .await?;
    threw(&ran, "stop")?;
    ensure(
        ran.seen.as_deref() == Some("array of 0"),
        format!(
            "the arm's `committed` is {:?}, not an empty array",
            ran.seen
        ),
    )
}

/// What a script returned or threw, its `seen` variable as it left it, and
/// the tables each block reported written directly.
struct Ran {
    out: std::result::Result<Dynamic, String>,
    seen: Option<String>,
    unprotected: Vec<Vec<String>>,
}

async fn eval(t: &TxTables, script: &'static str) -> Result<Ran> {
    eval_on(t, None, script).await
}

/// Runs `script` over the tables, and `other`'s as `c` and `d`, on a
/// blocking thread, since table calls block on the runtime. Each run gets
/// its own cell and atomic scope.
async fn eval_on(t: &TxTables, other: Option<&TxTables>, script: &'static str) -> Result<Ran> {
    let mut tables = vec![("a", t.a.clone()), ("b", t.b.clone())];
    if let Some(o) = other {
        tables.extend([("c", o.a.clone()), ("d", o.b.clone())]);
    }
    tokio::task::spawn_blocking(move || {
        let unprotected = Arc::new(Mutex::new(Vec::new()));
        let host = host(tables, unprotected.clone());
        let mut scope = Scope::new();
        let out = host
            .engine()
            .eval_with_scope::<Dynamic>(&mut scope, script)
            .map_err(|e| message(&e));
        let unprotected =
            std::mem::take(&mut *unprotected.lock().unwrap_or_else(PoisonError::into_inner));
        Ran {
            out,
            seen: scope.get_value::<String>("seen"),
            unprotected,
        }
    })
    .await
    .map_err(|e| error!(format!("the script's thread failed: {e}")))
}

fn host(tables: Vec<(&'static str, Vista)>, unprotected: Arc<Mutex<Vec<Vec<String>>>>) -> Host {
    let base: Arc<dyn VistaResolver> = Arc::new(move |name: &str| {
        tables
            .iter()
            .find(|(n, _)| *n == name)
            .map(|(_, vista)| vista.clone())
            .ok_or_else(|| error!("no such table", name = name))
    });
    let cell = ResolverCell::new(base);
    let terminals = Terminals::ReadWrite {
        limit: None,
        writes: Writes::Allowed,
    };
    let scope = AtomicScope::new(cell.clone())
        .with_limit(BLOCK_LIMIT)
        .with_arm_budget(ARM_BUDGET)
        .on_unprotected(Arc::new(move |tables| {
            let mut reported = unprotected.lock().unwrap_or_else(PoisonError::into_inner);
            reported.push(tables.to_vec());
        }));
    let clock = scope.clock();
    Host::builder(Limits::background())
        .vocab(DataVocab::new(terminals).with_resolver(Arc::new(cell)))
        .vocab(AtomicVocab::new(scope))
        .vocab_fn(move |engine| {
            engine.on_progress(move |_| {
                clock.should_stop(false).map(|stop| {
                    Dynamic::from(match stop {
                        Stop::Cancelled => "cancelled",
                        Stop::BlockTimeout => ATOMIC_BLOCK_TIMEOUT,
                        Stop::ArmTimeout => ATOMIC_ARM_TIMEOUT,
                    })
                })
            });
        })
        .build()
}

/// What the script threw, else the error's text.
fn message(err: &EvalAltResult) -> String {
    match err {
        EvalAltResult::ErrorRuntime(value, _) => value.to_string(),
        other => other.to_string(),
    }
}

fn threw(ran: &Ran, wanted: &str) -> Result<()> {
    match &ran.out {
        Err(e) if e == wanted => Ok(()),
        Err(e) => Err(error!(format!("the script throws `{e}`, not `{wanted}`"))),
        Ok(v) => Err(error!(format!("the script returns {v}, not `{wanted}`"))),
    }
}