improvise/crates/zset-ffi/tests/abi.rs

//! The ABI exercised through the `extern "C"` functions themselves, as a C caller would.
//! Evidence: `context/test-evidence/zset-ffi.md`.

use std::ffi::{CStr, c_void};
use std::ptr::{null, null_mut};
use zset_ffi::engine::Code;
use zset_ffi::*;

const OK: i32 = 0;

fn pair(k: &[u8], v: &[u8]) -> Vec<u8> {
    zset_ffi::engine::pair(k, v).unwrap()
}

fn zset(entries: &[(&[u8], i64)]) -> *mut ZsZSet {
    let mut z = null_mut();
    assert_eq!(unsafe { zs_zset_new(&mut z) }, OK);
    for (k, w) in entries {
        assert_eq!(unsafe { zs_zset_insert(z, k.as_ptr(), k.len(), *w) }, OK);
    }
    z
}

fn weight(z: *const ZsZSet, k: &[u8]) -> i64 {
    let mut w = 0;
    assert_eq!(
        unsafe { zs_zset_weight(z, k.as_ptr(), k.len(), &mut w) },
        OK
    );
    w
}

fn len(z: *const ZsZSet) -> usize {
    let mut n = 0;
    assert_eq!(unsafe { zs_zset_len(z, &mut n) }, OK);
    n
}

fn encoded(z: *const ZsZSet) -> Vec<u8> {
    let mut needed = 0;
    assert_eq!(
        unsafe { zs_zset_encode(z, null_mut(), 0, &mut needed) },
        if needed == 0 {
            OK
        } else {
            Code::BufferTooSmall as i32
        }
    );
    let mut buf = vec![0u8; needed];
    assert_eq!(
        unsafe { zs_zset_encode(z, buf.as_mut_ptr(), buf.len(), &mut needed) },
        OK
    );
    buf
}

fn last_error() -> String {
    unsafe { CStr::from_ptr(zs_last_error()) }
        .to_string_lossy()
        .into_owned()
}

struct C(*mut ZsCircuit);

impl C {
    fn new() -> C {
        let mut c = null_mut();
        assert_eq!(unsafe { zs_circuit_new(&mut c) }, OK);
        C(c)
    }
    fn input(&self) -> u32 {
        let mut s = 0;
        assert_eq!(unsafe { zs_circuit_input(self.0, &mut s) }, OK);
        s
    }
    fn sink(&self, s: u32) -> u32 {
        let mut o = 0;
        assert_eq!(unsafe { zs_circuit_output(self.0, s, &mut o) }, OK);
        o
    }
    fn push(&self, input: u32, entries: &[(&[u8], i64)]) {
        let z = zset(entries);
        assert_eq!(unsafe { zs_circuit_push(self.0, input, z) }, OK);
        unsafe { zs_zset_free(z) };
    }
    /// Steps and takes the sink as sorted `(key, weight)` pairs.
    fn run(&self, sink: u32) -> Vec<(Vec<u8>, i64)> {
        assert_eq!(unsafe { zs_circuit_step(self.0) }, OK);
        self.take(sink)
    }
    fn take(&self, sink: u32) -> Vec<(Vec<u8>, i64)> {
        let mut z = null_mut();
        assert_eq!(unsafe { zs_circuit_take(self.0, sink, &mut z) }, OK);
        let enc = encoded(z);
        unsafe { zs_zset_free(z) };
        decode_pairs(&enc)
    }
}

impl Drop for C {
    fn drop(&mut self) {
        unsafe { zs_circuit_free(self.0) }
    }
}

fn decode_pairs(mut b: &[u8]) -> Vec<(Vec<u8>, i64)> {
    let mut out = Vec::new();
    while !b.is_empty() {
        let n = u32::from_le_bytes(b[..4].try_into().unwrap()) as usize;
        let w = i64::from_le_bytes(b[4 + n..12 + n].try_into().unwrap());
        out.push((b[4..4 + n].to_vec(), w));
        b = &b[12 + n..];
    }
    out
}

fn one(entries: &[(&[u8], i64)]) -> Vec<(Vec<u8>, i64)> {
    entries.iter().map(|(k, w)| (k.to_vec(), *w)).collect()
}

fn le(n: i64) -> [u8; 8] {
    n.to_le_bytes()
}

#[test]
fn a_zset_adds_queries_and_frees() {
    let (a, b) = (
        zset(&[(b"x", 2), (b"y", -1)]),
        zset(&[(b"x", -2), (b"z", 5)]),
    );
    assert_eq!(unsafe { zs_zset_add(a, b) }, OK);
    assert_eq!(
        (weight(a, b"x"), weight(a, b"y"), weight(a, b"z"), len(a)),
        (0, -1, 5, 2)
    );
    assert_eq!(unsafe { zs_zset_neg(a) }, OK);
    assert_eq!(unsafe { zs_zset_scale(a, 3) }, OK);
    assert_eq!((weight(a, b"y"), weight(a, b"z")), (3, -15));
    assert_eq!(unsafe { zs_zset_sub(a, a) }, OK);
    assert_eq!(len(a), 0);
    for z in [a, b] {
        unsafe { zs_zset_free(z) };
    }
    unsafe { zs_zset_free(null_mut()) };
}

#[test]
fn keys_are_byte_strings_including_nul_and_empty() {
    let z = zset(&[(b"a\0b", 1), (b"", 2), (b"a", 3)]);
    assert_eq!(
        (weight(z, b"a\0b"), weight(z, b""), weight(z, b"a"), len(z)),
        (1, 2, 3, 3)
    );
    assert_eq!(unsafe { zs_zset_insert(z, null(), 0, 4) }, OK);
    assert_eq!(weight(z, b""), 6);
    unsafe { zs_zset_free(z) };
}

#[test]
fn distinct_clone_and_encoding_round_trip() {
    let z = zset(&[(b"k1", 5), (b"k2", -2), (b"k3", 1)]);
    let mut d = null_mut();
    assert_eq!(unsafe { zs_zset_distinct(z, &mut d) }, OK);
    assert_eq!(
        (weight(d, b"k1"), weight(d, b"k2"), weight(d, b"k3")),
        (1, 0, 1)
    );
    let mut copy = null_mut();
    assert_eq!(unsafe { zs_zset_clone(z, &mut copy) }, OK);
    let enc = encoded(z);
    let mut back = null_mut();
    assert_eq!(unsafe { zs_zset_new(&mut back) }, OK);
    assert_eq!(
        unsafe { zs_zset_decode_add(back, enc.as_ptr(), enc.len()) },
        OK
    );
    assert_eq!(encoded(back), encoded(copy));
    assert_eq!(encoded(back), enc);
    for h in [z, d, copy, back] {
        unsafe { zs_zset_free(h) };
    }
}

#[test]
fn a_small_buffer_reports_the_size_it_needs() {
    let z = zset(&[(b"abc", 1)]);
    let (mut needed, mut small) = (0, [0u8; 3]);
    assert_eq!(
        unsafe { zs_zset_encode(z, small.as_mut_ptr(), small.len(), &mut needed) },
        Code::BufferTooSmall as i32
    );
    assert_eq!(needed, 4 + 3 + 8);
    assert_eq!(small, [0; 3], "nothing written on failure");
    unsafe { zs_zset_free(z) };
}

#[test]
fn a_bad_encoding_is_refused_and_adds_nothing() {
    let z = zset(&[(b"a", 1)]);
    let mut enc = encoded(z);
    enc.push(7);
    assert_eq!(
        unsafe { zs_zset_decode_add(z, enc.as_ptr(), enc.len()) },
        Code::BadValue as i32
    );
    assert_eq!((weight(z, b"a"), len(z)), (1, 1));
    unsafe { zs_zset_free(z) };
}

#[test]
fn overflow_is_an_error_code_and_leaves_the_set_unchanged() {
    let (a, b) = (
        zset(&[(b"k", i64::MAX), (b"j", 1)]),
        zset(&[(b"j", 1), (b"k", 1)]),
    );
    assert_eq!(unsafe { zs_zset_add(a, b) }, Code::Overflow as i32);
    assert_eq!(
        (weight(a, b"k"), weight(a, b"j")),
        (i64::MAX, 1),
        "a is unchanged, even for the key before the overflow"
    );
    assert!(last_error().contains("weight overflow"), "{}", last_error());
    assert_eq!(unsafe { zs_zset_scale(a, 2) }, Code::Overflow as i32);
    assert_eq!(
        unsafe { zs_zset_insert(a, b"k".as_ptr(), 1, 1) },
        Code::Overflow as i32
    );
    let m = zset(&[(b"k", i64::MIN)]);
    assert_eq!(unsafe { zs_zset_neg(m) }, Code::Overflow as i32);
    assert_eq!(weight(m, b"k"), i64::MIN);
    assert_eq!(unsafe { zs_zset_sub(a, a) }, OK);
    for z in [a, b, m] {
        unsafe { zs_zset_free(z) };
    }
}

#[test]
fn a_null_argument_is_a_code_not_a_crash() {
    assert_eq!(
        unsafe { zs_zset_new(null_mut()) },
        Code::NullArgument as i32
    );
    assert_eq!(
        unsafe { zs_zset_len(null(), &mut 0) },
        Code::NullArgument as i32
    );
    let z = zset(&[]);
    assert_eq!(
        unsafe { zs_zset_insert(z, null(), 3, 1) },
        Code::NullArgument as i32
    );
    assert_eq!(unsafe { zs_zset_add(z, null()) }, Code::NullArgument as i32);
    assert_eq!(
        unsafe { zs_circuit_filter(null_mut(), 0, None, null_mut(), &mut 0) },
        Code::NullArgument as i32
    );
    unsafe { zs_zset_free(z) };
    assert_ne!(zs_abi_version(), 0);
}

#[test]
fn a_circuit_runs_distinct_and_integrate() {
    let c = C::new();
    let i = c.input();
    let (mut d, mut int) = (0, 0);
    assert_eq!(unsafe { zs_circuit_distinct(c.0, i, &mut d) }, OK);
    assert_eq!(unsafe { zs_circuit_integrate(c.0, d, &mut int) }, OK);
    let (sd, sint) = (c.sink(d), c.sink(int));
    c.push(i, &[(b"a", 3), (b"b", -1)]);
    assert_eq!(unsafe { zs_circuit_step(c.0) }, OK);
    assert_eq!(c.take(sd), one(&[(b"a", 1)]));
    assert_eq!(c.take(sint), one(&[(b"a", 1)]));
    c.push(i, &[(b"a", -3), (b"b", 1)]);
    assert_eq!(c.run(sd), one(&[(b"a", -1)]));
}

extern "C" fn upper(
    _: *mut c_void,
    k: *const u8,
    n: usize,
    emit: engine::Emit,
    e: *mut c_void,
) -> i32 {
    let up: Vec<u8> = unsafe { std::slice::from_raw_parts(k, n) }.to_ascii_uppercase();
    emit(e, up.as_ptr(), up.len());
    emit(e, b"!".as_ptr(), 1);
    0
}

extern "C" fn even_len(_: *mut c_void, _: *const u8, n: usize) -> i32 {
    i32::from(n.is_multiple_of(2))
}

extern "C" fn first_byte(
    _: *mut c_void,
    k: *const u8,
    n: usize,
    emit: engine::Emit,
    e: *mut c_void,
) -> i32 {
    emit(e, k, n.min(1));
    0
}

#[test]
fn callbacks_filter_and_flat_map() {
    let c = C::new();
    let i = c.input();
    let (mut f, mut m) = (0, 0);
    assert_eq!(
        unsafe { zs_circuit_filter(c.0, i, Some(even_len), null_mut(), &mut f) },
        OK
    );
    assert_eq!(
        unsafe { zs_circuit_flat_map(c.0, f, Some(upper), null_mut(), &mut m) },
        OK
    );
    let s = c.sink(m);
    c.push(i, &[(b"ab", 2), (b"abc", 5), (b"cd", -1)]);
    assert_eq!(c.run(s), one(&[(b"!", 1), (b"AB", 2), (b"CD", -1)]));
}

#[test]
fn index_by_join_and_aggregate_follow_the_pair_encoding() {
    let c = C::new();
    let (orders, prices) = (c.input(), c.input());
    let (mut by_item, mut joined) = (0, 0);
    assert_eq!(
        unsafe { zs_circuit_index_by(c.0, orders, Some(first_byte), null_mut(), &mut by_item) },
        OK
    );
    assert_eq!(
        unsafe { zs_circuit_join(c.0, by_item, prices, &mut joined) },
        OK
    );
    let sink = c.sink(joined);
    c.push(orders, &[(b"apple", 1)]);
    c.push(prices, &[(&pair(b"a", b"3"), 1)]);
    let want = pair(b"a", &pair(b"apple", b"3"));
    assert_eq!(c.run(sink), one(&[(&want, 1)]));
    c.push(prices, &[(&pair(b"a", b"3"), -1)]);
    assert_eq!(c.run(sink), one(&[(&want, -1)]));
}

#[test]
fn aggregates_count_sum_min_and_max() {
    let c = C::new();
    let i = c.input();
    let outs: Vec<u32> = (0..4)
        .map(|kind| {
            let mut o = 0;
            assert_eq!(unsafe { zs_circuit_aggregate(c.0, i, kind, &mut o) }, OK);
            c.sink(o)
        })
        .collect();
    c.push(
        i,
        &[
            (&pair(b"g", &le(5)), 1),
            (&pair(b"g", &le(2)), 2),
            (&pair(b"h", &le(9)), 1),
        ],
    );
    assert_eq!(unsafe { zs_circuit_step(c.0) }, OK);
    let val = |k: &[u8], n: i64| (pair(k, &le(n)), 1);
    let want = [
        [val(b"g", 3), val(b"h", 1)],
        [val(b"g", 9), val(b"h", 9)],
        [val(b"g", 2), val(b"h", 9)],
        [val(b"g", 5), val(b"h", 9)],
    ];
    for (sink, want) in outs.iter().zip(want) {
        let got = c.take(*sink);
        let want: Vec<(Vec<u8>, i64)> = want.into_iter().collect();
        let mut sorted = want.clone();
        sorted.sort();
        assert_eq!(got, sorted, "kind of sink {sink}");
    }
}

#[test]
fn a_weight_overflow_in_a_step_is_a_code_and_poisons_the_circuit() {
    let c = C::new();
    let i = c.input();
    let (mut int, mut again) = (0, 0);
    assert_eq!(unsafe { zs_circuit_integrate(c.0, i, &mut int) }, OK);
    c.push(i, &[(b"k", i64::MAX)]);
    assert_eq!(unsafe { zs_circuit_step(c.0) }, OK);
    c.push(i, &[(b"k", 1)]);
    assert_eq!(unsafe { zs_circuit_step(c.0) }, Code::Overflow as i32);
    assert!(last_error().contains("weight overflow"));
    assert_eq!(unsafe { zs_circuit_step(c.0) }, Code::Poisoned as i32);
    assert_eq!(
        unsafe { zs_circuit_distinct(c.0, i, &mut again) },
        Code::Poisoned as i32
    );
}

#[test]
fn a_malformed_pair_and_an_unknown_id_are_codes() {
    let c = C::new();
    let i = c.input();
    let mut j = 0;
    assert_eq!(
        unsafe { zs_circuit_join(c.0, i, 99, &mut j) },
        Code::BadHandle as i32
    );
    assert_eq!(unsafe { zs_circuit_join(c.0, i, i, &mut j) }, OK);
    c.push(i, &[(b"\xff\xff\xff\xff", 1)]);
    assert_eq!(unsafe { zs_circuit_step(c.0) }, Code::BadValue as i32);
    assert_eq!(unsafe { zs_circuit_step(c.0) }, Code::Poisoned as i32);
    let mut a = 0;
    let c2 = C::new();
    assert_eq!(
        unsafe { zs_circuit_aggregate(c2.0, 0, 9, &mut a) },
        Code::BadValue as i32
    );
    let mut z = null_mut();
    assert_eq!(
        unsafe { zs_circuit_take(c2.0, 5, &mut z) },
        Code::BadHandle as i32
    );
    let empty = zset(&[]);
    assert_eq!(
        unsafe { zs_circuit_push(c2.0, 0, empty) },
        Code::BadHandle as i32
    );
    unsafe { zs_zset_free(empty) };
}

extern "C" fn out_of_time(ctx: *mut c_void) -> i32 {
    let calls = unsafe { &mut *(ctx as *mut u32) };
    *calls += 1;
    1
}

#[test]
fn a_rebuild_resumes_across_calls_and_matches_one_step() {
    let keys: Vec<Vec<u8>> = (0u8..9).map(|n| vec![b'k', n]).collect();
    let entries: Vec<(&[u8], i64)> = keys.iter().map(|k| (k.as_slice(), 1)).collect();
    let snapshot = zset(&entries);

    let whole = C::new();
    let wi = whole.input();
    let mut wd = 0;
    assert_eq!(unsafe { zs_circuit_distinct(whole.0, wi, &mut wd) }, OK);
    let ws = whole.sink(wd);
    assert_eq!(unsafe { zs_circuit_push(whole.0, wi, snapshot) }, OK);
    let want = whole.run(ws);

    let c = C::new();
    let i = c.input();
    let mut d = 0;
    assert_eq!(unsafe { zs_circuit_distinct(c.0, i, &mut d) }, OK);
    let s = c.sink(d);
    let mut job = null_mut();
    assert_eq!(unsafe { zs_rebuild_new(c.0, i, snapshot, 2, &mut job) }, OK);
    let (mut calls, mut got, mut done) = (0u32, Vec::new(), 0);
    for _ in 0..20 {
        assert_eq!(
            unsafe {
                zs_rebuild_step(
                    job,
                    c.0,
                    Some(out_of_time),
                    (&mut calls as *mut u32).cast(),
                    &mut done,
                )
            },
            OK
        );
        got.extend(c.take(s));
        if done == 1 {
            break;
        }
    }
    assert_eq!(done, 1, "the job finished within its bound");
    assert_eq!(
        calls, 4,
        "ceil(9 / 2) = 5 chunks; the last is Done before any check"
    );
    got.sort();
    assert_eq!(got, want);
    unsafe { zs_rebuild_free(job) };
    unsafe { zs_zset_free(snapshot) };
}

#[test]
fn a_rebuild_without_a_deadline_runs_to_completion() {
    let snapshot = zset(&[(b"a", 1), (b"b", 1), (b"c", 1)]);
    let c = C::new();
    let i = c.input();
    let s = c.sink(i);
    let mut job = null_mut();
    let mut done = 0;
    assert_eq!(unsafe { zs_rebuild_new(c.0, i, snapshot, 1, &mut job) }, OK);
    assert_eq!(
        unsafe { zs_rebuild_step(job, c.0, None, null_mut(), &mut done) },
        OK
    );
    assert_eq!(done, 1);
    assert_eq!(c.take(s).len(), 3);
    unsafe { zs_rebuild_free(job) };
    unsafe { zs_zset_free(snapshot) };
}

#[test]
fn min_and_max_read_only_values_of_positive_weight() {
    let c = C::new();
    let i = c.input();
    let (mut lo, mut hi) = (0, 0);
    assert_eq!(unsafe { zs_circuit_aggregate(c.0, i, 2, &mut lo) }, OK);
    assert_eq!(unsafe { zs_circuit_aggregate(c.0, i, 3, &mut hi) }, OK);
    let (slo, shi) = (c.sink(lo), c.sink(hi));
    c.push(
        i,
        &[
            (&pair(b"g", &le(5)), 1),
            (&pair(b"g", &le(1)), -1),
            (&pair(b"g", &le(9)), -1),
        ],
    );
    assert_eq!(unsafe { zs_circuit_step(c.0) }, OK);
    assert_eq!(c.take(slo), one(&[(&pair(b"g", &le(5)), 1)]));
    assert_eq!(c.take(shi), one(&[(&pair(b"g", &le(5)), 1)]));
}

#[test]
fn an_index_by_callback_that_emits_nothing_is_a_bad_value() {
    extern "C" fn nothing(
        _: *mut c_void,
        _: *const u8,
        _: usize,
        _: engine::Emit,
        _: *mut c_void,
    ) -> i32 {
        0
    }
    let c = C::new();
    let i = c.input();
    let mut o = 0;
    assert_eq!(
        unsafe { zs_circuit_index_by(c.0, i, Some(nothing), null_mut(), &mut o) },
        OK
    );
    c.push(i, &[(b"x", 1)]);
    assert_eq!(unsafe { zs_circuit_step(c.0) }, Code::BadValue as i32);
    assert!(last_error().contains("exactly one key"), "{}", last_error());
}

#[test]
fn a_chunked_rebuild_through_delay_is_one_tick() {
    let snapshot = zset(&[(b"a", 1), (b"b", 1)]);
    let c = C::new();
    let i = c.input();
    let mut d = 0;
    assert_eq!(unsafe { zs_circuit_delay(c.0, i, &mut d) }, OK);
    let s = c.sink(d);
    c.push(i, &[(b"p", 1)]);
    assert_eq!(unsafe { zs_circuit_step(c.0) }, OK);
    c.take(s);
    let mut job = null_mut();
    assert_eq!(unsafe { zs_rebuild_new(c.0, i, snapshot, 1, &mut job) }, OK);
    let (mut calls, mut done) = (0u32, 0);
    let ctx = (&mut calls as *mut u32).cast();
    assert_eq!(
        unsafe { zs_rebuild_step(job, c.0, Some(out_of_time), ctx, &mut done) },
        OK
    );
    assert_eq!(done, 0);
    assert_eq!(
        unsafe { zs_rebuild_step(job, c.0, Some(out_of_time), ctx, &mut done) },
        OK
    );
    assert_eq!(
        done, 1,
        "the last chunk is Done though the deadline expired"
    );
    assert_eq!(c.take(s), one(&[(b"p", 1)]));
    unsafe { zs_rebuild_free(job) };
    unsafe { zs_zset_free(snapshot) };
}

#[test]
fn a_set_added_to_or_subtracted_from_itself_is_defined() {
    let a = zset(&[(b"x", 2), (b"y", -3)]);
    assert_eq!(unsafe { zs_zset_add(a, a) }, OK);
    assert_eq!((weight(a, b"x"), weight(a, b"y")), (4, -6));
    assert_eq!(unsafe { zs_zset_sub(a, a) }, OK);
    assert_eq!(len(a), 0);
    let m = zset(&[(b"k", i64::MAX)]);
    assert_eq!(unsafe { zs_zset_add(m, m) }, Code::Overflow as i32);
    assert_eq!(
        weight(m, b"k"),
        i64::MAX,
        "the failed self-add changed nothing"
    );
    unsafe { zs_zset_free(a) };
    unsafe { zs_zset_free(m) };
}

#[test]
fn a_null_buffer_is_refused_before_it_is_written() {
    let z = zset(&[(b"abc", 1)]);
    let mut needed = 0;
    assert_eq!(
        unsafe { zs_zset_encode(z, null_mut(), 100, &mut needed) },
        Code::NullArgument as i32
    );
    assert_eq!(
        unsafe { zs_zset_encode(z, null_mut(), 0, &mut needed) },
        Code::BufferTooSmall as i32,
        "a null buffer still probes the size"
    );
    assert_eq!(needed, 15);
    let empty = zset(&[]);
    assert_eq!(
        unsafe { zs_zset_encode(empty, null_mut(), 0, &mut needed) },
        OK
    );
    unsafe { zs_zset_free(z) };
    unsafe { zs_zset_free(empty) };
}

#[test]
fn a_null_out_pointer_is_refused_before_any_state_changes() {
    let c = C::new();
    let i = c.input();
    assert_eq!(
        unsafe { zs_circuit_input(c.0, null_mut()) },
        Code::NullArgument as i32
    );
    assert_eq!(c.input(), i + 1, "the refused call added no input");
    let s = c.sink(i);
    c.push(i, &[(b"k", 1)]);
    assert_eq!(unsafe { zs_circuit_step(c.0) }, OK);
    assert_eq!(
        unsafe { zs_circuit_take(c.0, s, null_mut()) },
        Code::NullArgument as i32
    );
    assert_eq!(
        c.take(s),
        one(&[(b"k", 1)]),
        "the refused take left the sink alone"
    );
    assert_eq!(
        unsafe { zs_zset_new(null_mut()) },
        Code::NullArgument as i32
    );
    let snapshot = zset(&[(b"a", 1), (b"b", 1)]);
    let mut job = null_mut();
    assert_eq!(unsafe { zs_rebuild_new(c.0, i, snapshot, 1, &mut job) }, OK);
    assert_eq!(
        unsafe { zs_rebuild_step(job, c.0, None, null_mut(), null_mut()) },
        Code::NullArgument as i32
    );
    let mut done = 0;
    assert_eq!(
        unsafe { zs_rebuild_step(job, c.0, None, null_mut(), &mut done) },
        OK
    );
    assert_eq!(c.take(s).len(), 2, "the refused step consumed nothing");
    unsafe { zs_rebuild_free(job) };
    unsafe { zs_zset_free(snapshot) };
}

#[test]
fn a_rebuild_is_stepped_only_on_its_own_circuit() {
    let (c1, c2) = (C::new(), C::new());
    let (i1, i2) = (c1.input(), c2.input());
    let (s1, s2) = (c1.sink(i1), c2.sink(i2));
    let snapshot = zset(&[(b"a", 1)]);
    let mut job = null_mut();
    assert_eq!(
        unsafe { zs_rebuild_new(c1.0, i1, snapshot, 1, &mut job) },
        OK
    );
    let mut done = 0;
    assert_eq!(
        unsafe { zs_rebuild_step(job, c2.0, None, null_mut(), &mut done) },
        Code::BadHandle as i32
    );
    assert_eq!((c2.take(s2).len(), c1.take(s1).len()), (0, 0));
    assert_eq!(
        unsafe { zs_rebuild_step(job, c1.0, None, null_mut(), &mut done) },
        OK
    );
    assert_eq!(c1.take(s1), one(&[(b"a", 1)]));
    unsafe { zs_rebuild_free(job) };
    unsafe { zs_zset_free(snapshot) };
}

#[test]
fn stream_ids_and_sink_ids_are_not_interchangeable() {
    let c = C::new();
    let i = c.input();
    let s = c.sink(i);
    let mut z = null_mut();
    assert_eq!(
        unsafe { zs_circuit_take(c.0, i, &mut z) },
        Code::BadHandle as i32
    );
    let mut o = 0;
    assert_eq!(
        unsafe { zs_circuit_delay(c.0, s, &mut o) },
        Code::BadHandle as i32
    );
    assert_eq!(
        unsafe { zs_circuit_output(c.0, s, &mut o) },
        Code::BadHandle as i32
    );
    let empty = zset(&[]);
    assert_eq!(
        unsafe { zs_circuit_push(c.0, s, empty) },
        Code::BadHandle as i32
    );
    unsafe { zs_zset_free(empty) };
}

extern "C" fn fails(_: *mut c_void, _: *const u8, _: usize) -> i32 {
    -1
}

extern "C" fn flat_fails(
    _: *mut c_void,
    _: *const u8,
    _: usize,
    _: engine::Emit,
    _: *mut c_void,
) -> i32 {
    7
}

extern "C" fn expire_fails(_: *mut c_void) -> i32 {
    -1
}

#[test]
fn a_callback_that_reports_failure_fails_the_step_and_kills_the_circuit() {
    let c = C::new();
    let i = c.input();
    let (mut f, mut m) = (0, 0);
    assert_eq!(
        unsafe { zs_circuit_filter(c.0, i, Some(fails), null_mut(), &mut f) },
        OK
    );
    c.push(i, &[(b"k", 1)]);
    assert_eq!(unsafe { zs_circuit_step(c.0) }, Code::Callback as i32);
    assert_eq!(unsafe { zs_circuit_step(c.0) }, Code::Poisoned as i32);
    let d = C::new();
    let j = d.input();
    assert_eq!(
        unsafe { zs_circuit_flat_map(d.0, j, Some(flat_fails), null_mut(), &mut m) },
        OK
    );
    d.push(j, &[(b"k", 1)]);
    assert_eq!(unsafe { zs_circuit_step(d.0) }, Code::Callback as i32);
    let e = C::new();
    let k = e.input();
    let snapshot = zset(&[(b"a", 1), (b"b", 1)]);
    let mut job = null_mut();
    assert_eq!(unsafe { zs_rebuild_new(e.0, k, snapshot, 1, &mut job) }, OK);
    let mut done = 0;
    assert_eq!(
        unsafe { zs_rebuild_step(job, e.0, Some(expire_fails), null_mut(), &mut done) },
        Code::Callback as i32
    );
    unsafe { zs_rebuild_free(job) };
    unsafe { zs_zset_free(snapshot) };
}

extern "C" fn reenters(ctx: *mut c_void, _: *const u8, _: usize) -> i32 {
    let (c, seen) = unsafe { &mut *(ctx as *mut (*mut ZsCircuit, i32)) };
    let mut id = 0;
    *seen = unsafe { zs_circuit_input(*c, &mut id) };
    1
}

#[test]
fn a_callback_that_calls_back_into_its_circuit_gets_poisoned_and_changes_nothing() {
    let c = C::new();
    let i = c.input();
    let mut state: (*mut ZsCircuit, i32) = (c.0, 0);
    let mut f = 0;
    let ctx = (&mut state as *mut (*mut ZsCircuit, i32)).cast();
    assert_eq!(
        unsafe { zs_circuit_filter(c.0, i, Some(reenters), ctx, &mut f) },
        OK
    );
    c.push(i, &[(b"k", 1)]);
    assert_eq!(
        unsafe { zs_circuit_step(c.0) },
        OK,
        "the outer step completes"
    );
    assert_eq!(state.1, Code::Poisoned as i32);
    assert_eq!(c.input(), f + 1, "the re-entrant call added no input");
}

#[test]
fn a_count_or_sum_that_cancels_to_zero_keeps_its_row() {
    let c = C::new();
    let i = c.input();
    let (mut n, mut t) = (0, 0);
    assert_eq!(unsafe { zs_circuit_aggregate(c.0, i, 0, &mut n) }, OK);
    assert_eq!(unsafe { zs_circuit_aggregate(c.0, i, 1, &mut t) }, OK);
    let (sn, st) = (c.sink(n), c.sink(t));
    c.push(
        i,
        &[
            (&pair(b"g", &le(5)), 1),
            (&pair(b"g", &le(-5)), 1),
            (&pair(b"h", &le(1)), -1),
            (&pair(b"h", &le(2)), 1),
        ],
    );
    assert_eq!(unsafe { zs_circuit_step(c.0) }, OK);
    assert_eq!(
        c.take(st),
        one(&[(&pair(b"g", &le(0)), 1), (&pair(b"h", &le(1)), 1)])
    );
    assert_eq!(
        c.take(sn),
        one(&[(&pair(b"g", &le(2)), 1), (&pair(b"h", &le(0)), 1)])
    );
}

#[cfg(target_pointer_width = "64")]
#[test]
fn a_length_of_4_gib_or_more_is_refused_not_truncated() {
    assert_eq!(engine::len32(u32::MAX as usize), Ok(u32::MAX));
    assert_eq!(
        engine::len32(1usize << 32).unwrap_err().code,
        Code::BadValue
    );
}

#[test]
fn an_index_by_callback_that_fails_reports_the_callback_code_not_a_later_fault() {
    extern "C" fn fails_after_emitting(
        _: *mut c_void,
        _: *const u8,
        _: usize,
        _: engine::Emit,
        _: *mut c_void,
    ) -> i32 {
        3
    }
    let c = C::new();
    let i = c.input();
    let mut o = 0;
    assert_eq!(
        unsafe { zs_circuit_index_by(c.0, i, Some(fails_after_emitting), null_mut(), &mut o) },
        OK
    );
    c.push(i, &[(b"x", 1)]);
    assert_eq!(unsafe { zs_circuit_step(c.0) }, Code::Callback as i32);
}

struct Reenter {
    c: *mut ZsCircuit,
    job: *mut ZsRebuild,
    seen: Vec<i32>,
}

extern "C" fn expired_reenters(ctx: *mut c_void) -> i32 {
    let r = unsafe { &mut *(ctx as *mut Reenter) };
    let mut done = 0;
    r.seen
        .push(unsafe { zs_rebuild_step(r.job, r.c, None, null_mut(), &mut done) });
    unsafe { zs_rebuild_free(r.job) };
    unsafe { zs_circuit_free(r.c) };
    1
}

#[test]
fn a_callback_cannot_step_or_free_the_job_or_circuit_it_runs_under() {
    let snapshot = zset(&[(b"a", 1), (b"b", 1), (b"c", 1)]);
    let c = C::new();
    let i = c.input();
    let s = c.sink(i);
    let mut job = null_mut();
    assert_eq!(unsafe { zs_rebuild_new(c.0, i, snapshot, 1, &mut job) }, OK);
    let mut r = Reenter {
        c: c.0,
        job,
        seen: Vec::new(),
    };
    let mut done = 0;
    let ctx = (&mut r as *mut Reenter).cast();
    assert_eq!(
        unsafe { zs_rebuild_step(job, c.0, Some(expired_reenters), ctx, &mut done) },
        OK
    );
    assert_eq!(
        r.seen,
        vec![Code::Poisoned as i32],
        "stepping from inside is refused"
    );
    assert_eq!(done, 0);
    assert_eq!(
        unsafe { zs_rebuild_step(job, c.0, None, null_mut(), &mut done) },
        OK
    );
    assert_eq!(
        (done, c.take(s).len()),
        (1, 3),
        "the job and circuit survived the frees"
    );
    unsafe { zs_rebuild_free(job) };
    unsafe { zs_zset_free(snapshot) };
}

#[test]
fn a_circuit_freed_and_another_made_does_not_inherit_its_jobs() {
    let snapshot = zset(&[(b"a", 1), (b"b", 1)]);
    let c1 = C::new();
    let i1 = c1.input();
    let mut job = null_mut();
    assert_eq!(
        unsafe { zs_rebuild_new(c1.0, i1, snapshot, 1, &mut job) },
        OK
    );
    drop(c1);
    for _ in 0..64 {
        let c2 = C::new();
        let _ = c2.input();
        let mut done = 0;
        assert_eq!(
            unsafe { zs_rebuild_step(job, c2.0, None, null_mut(), &mut done) },
            Code::BadHandle as i32,
            "a new circuit, even at the old address, is not the job's"
        );
    }
    unsafe { zs_rebuild_free(job) };
    unsafe { zs_zset_free(snapshot) };
}

#[test]
fn a_host_step_between_rebuild_calls_is_interrupted_not_a_poison() {
    let snapshot = zset(&[(b"a", 1), (b"b", 1)]);
    let c = C::new();
    let i = c.input();
    let s = c.sink(i);
    let mut job = null_mut();
    assert_eq!(unsafe { zs_rebuild_new(c.0, i, snapshot, 1, &mut job) }, OK);
    let (mut calls, mut done) = (0u32, 0);
    let ctx = (&mut calls as *mut u32).cast();
    assert_eq!(
        unsafe { zs_rebuild_step(job, c.0, Some(out_of_time), ctx, &mut done) },
        OK
    );
    assert_eq!(unsafe { zs_circuit_step(c.0) }, OK);
    assert_eq!(
        unsafe { zs_rebuild_step(job, c.0, None, null_mut(), &mut done) },
        Code::Interrupted as i32
    );
    assert!(last_error().contains("interrupted"), "{}", last_error());
    assert_eq!(
        unsafe { zs_circuit_step(c.0) },
        OK,
        "the circuit is still usable"
    );
    let _ = c.take(s);
    unsafe { zs_rebuild_free(job) };
    unsafe { zs_zset_free(snapshot) };
}

/// An integrate over `k` held at `i64::MAX` (taken), and its sink.
fn at_max() -> (C, u32, u32) {
    let c = C::new();
    let i = c.input();
    let mut int = 0;
    assert_eq!(unsafe { zs_circuit_integrate(c.0, i, &mut int) }, OK);
    let s = c.sink(int);
    c.push(i, &[(b"k", i64::MAX)]);
    assert_eq!(c.run(s), one(&[(b"k", i64::MAX)]));
    (c, i, s)
}

#[test]
fn a_rollback_after_a_failed_step_revives_the_circuit_in_its_old_state() {
    let (c, i, s) = at_max();
    assert_eq!(unsafe { zs_circuit_begin(c.0) }, OK);
    c.push(i, &[(b"a", 1), (b"k", 1)]);
    assert_eq!(unsafe { zs_circuit_step(c.0) }, Code::Overflow as i32);
    assert_eq!(unsafe { zs_circuit_step(c.0) }, Code::Poisoned as i32);
    assert_eq!(
        unsafe { zs_circuit_commit(c.0) },
        Code::Poisoned as i32,
        "a failed step cannot be committed"
    );
    assert_eq!(unsafe { zs_circuit_rollback(c.0) }, OK);
    c.push(i, &[(b"k", -1)]);
    assert_eq!(
        c.run(s),
        one(&[(b"k", i64::MAX - 1)]),
        "[rollback] neither `a` nor the failed +1 is in the integral"
    );
}

#[test]
fn a_committed_transaction_keeps_its_steps_and_a_new_one_may_follow() {
    let (c, i, s) = at_max();
    assert_eq!(unsafe { zs_circuit_begin(c.0) }, OK);
    c.push(i, &[(b"k", -1)]);
    assert_eq!(unsafe { zs_circuit_step(c.0) }, OK);
    assert_eq!(unsafe { zs_circuit_commit(c.0) }, OK);
    assert_eq!(c.take(s), one(&[(b"k", i64::MAX - 1)]));
    assert_eq!(unsafe { zs_circuit_begin(c.0) }, OK);
    assert_eq!(unsafe { zs_circuit_rollback(c.0) }, OK);
    c.push(i, &[(b"k", -1)]);
    assert_eq!(c.run(s), one(&[(b"k", i64::MAX - 2)]));
}

#[test]
fn a_failed_callback_step_is_rolled_back_too() {
    let c = C::new();
    let i = c.input();
    let (mut f, mut int) = (0, 0);
    assert_eq!(unsafe { zs_circuit_integrate(c.0, i, &mut int) }, OK);
    let s = c.sink(int);
    assert_eq!(
        unsafe { zs_circuit_filter(c.0, i, Some(fails), null_mut(), &mut f) },
        OK
    );
    assert_eq!(unsafe { zs_circuit_begin(c.0) }, OK);
    c.push(i, &[(b"k", 1)]);
    assert_eq!(unsafe { zs_circuit_step(c.0) }, Code::Callback as i32);
    assert_eq!(unsafe { zs_circuit_rollback(c.0) }, OK);
    assert_eq!(unsafe { zs_circuit_begin(c.0) }, OK, "usable again");
    assert_eq!(unsafe { zs_circuit_commit(c.0) }, OK);
    assert_eq!(
        c.take(s),
        vec![],
        "[callback] the failed step left no output"
    );
}

#[test]
fn transaction_misuse_is_a_code_and_changes_nothing() {
    let (c, i, s) = at_max();
    let t = Code::Transaction as i32;
    assert_eq!(unsafe { zs_circuit_commit(c.0) }, t);
    assert_eq!(unsafe { zs_circuit_rollback(c.0) }, t);
    assert_eq!(unsafe { zs_circuit_begin(c.0) }, OK);
    assert_eq!(unsafe { zs_circuit_begin(c.0) }, t);
    let mut z = null_mut();
    assert_eq!(unsafe { zs_circuit_take(c.0, s, &mut z) }, t);
    assert!(z.is_null(), "nothing written");
    assert!(last_error().contains("roll back"), "{}", last_error());
    assert_eq!(unsafe { zs_circuit_commit(c.0) }, OK);
    // A poison from outside any transaction is not undone by a rollback.
    c.push(i, &[(b"k", 1)]);
    assert_eq!(unsafe { zs_circuit_step(c.0) }, Code::Overflow as i32);
    assert_eq!(unsafe { zs_circuit_rollback(c.0) }, t);
    assert_eq!(unsafe { zs_circuit_step(c.0) }, Code::Poisoned as i32);
}