grit-core 0.2.4

//! Bi-temporal invariants: event-time validity filtering, system-time
//! ("what did I believe at t?") filtering, and MIN-convergence of concurrent
//! invalidations.

use std::sync::Arc;

use grit_core::{Budget, GraphOp, Grit, ManualClock, Options, Query, Traversal};
use serde_json::json;
use uuid::Uuid;

struct Fixture {
    _dir: tempfile::TempDir,
    clock: Arc<ManualClock>,
    g: Grit,
}

fn fixture(start_ms: i64) -> Fixture {
    let dir = tempfile::tempdir().unwrap();
    let clock = Arc::new(ManualClock::new(start_ms));
    let g = Grit::open(
        dir.path().join("m.db"),
        Options::new("dev").clock(clock.clone()),
    )
    .unwrap();
    Fixture {
        _dir: dir,
        clock,
        g,
    }
}

fn node(g: &Grit, name: &str) -> Uuid {
    let id = g.new_id();
    g.apply(GraphOp::AddNode {
        id,
        kind: "entity".into(),
        name: name.into(),
        summary: String::new(),
        attrs: json!({}),
        group_id: String::new(),
    })
    .unwrap();
    id
}

/// Alice works at Acme from t=2000 (event time); at t=10_000 we learn the job
/// ended at t=8000. All four quadrants of the bi-temporal square must answer
/// correctly.
#[test]
fn event_time_and_system_time_are_independent() {
    let f = fixture(1_000);
    let alice = node(&f.g, "Alice");
    let acme = node(&f.g, "Acme");

    f.clock.set_ms(2_000);
    let job = f.g.new_id();
    f.g.apply(GraphOp::AddEdge {
        id: job,
        src: alice,
        dst: acme,
        rel: "WORKS_AT".into(),
        fact: "Alice works at Acme".into(),
        attrs: json!({}),
        group_id: String::new(),
        valid_at: Some(2_000),
        invalid_at: None,
    })
    .unwrap();

    f.clock.set_ms(10_000);
    f.g.apply(GraphOp::InvalidateEdge {
        edge_id: job,
        invalid_at: 8_000,
    })
    .unwrap();

    let edges_at = |as_of: i64, as_at: i64| {
        f.g.traverse(&[alice], &Traversal::default().as_of(as_of).as_at(as_at))
            .unwrap()
            .edges
            .len()
    };

    // Fact true at 5000, and we believed it back then too.
    assert_eq!(edges_at(5_000, 5_000), 1);
    // Fact true at 5000, believed today (we now know it ended at 8000 — but
    // 5000 < 8000, so it still holds *for that instant*).
    assert_eq!(edges_at(5_000, 20_000), 1);
    // Fact no longer true at 9000, believed today.
    assert_eq!(edges_at(9_000, 20_000), 0);
    // The payoff query: "back at t=5000, did we think the job would still
    // hold at t=9000?" The invalidation was only recorded at t=10_000, so at
    // belief-time 5000 the interval was [2000, ∞) — yes, we believed it.
    // Invalidations are belief-versioned (edge_invalidations.recorded_at).
    assert_eq!(edges_at(9_000, 5_000), 1);
    // Before the edge was recorded, nothing was believed.
    assert_eq!(edges_at(5_000, 1_500), 0);
    // Before the fact's valid_at, it was not yet true.
    assert_eq!(edges_at(1_800, 20_000), 0);
}

#[test]
fn concurrent_invalidations_converge_to_min() {
    let f = fixture(1_000);
    let a = node(&f.g, "A");
    let b = node(&f.g, "B");
    let edge = f.g.new_id();
    f.g.apply(GraphOp::AddEdge {
        id: edge,
        src: a,
        dst: b,
        rel: "R".into(),
        fact: String::new(),
        attrs: json!({}),
        group_id: String::new(),
        valid_at: None,
        invalid_at: None,
    })
    .unwrap();

    // Later invalidation lands first, earlier one second: MIN wins.
    f.g.apply(GraphOp::InvalidateEdge {
        edge_id: edge,
        invalid_at: 9_000,
    })
    .unwrap();
    assert_eq!(f.g.edge(edge).unwrap().unwrap().invalid_at, Some(9_000));
    f.g.apply(GraphOp::InvalidateEdge {
        edge_id: edge,
        invalid_at: 3_000,
    })
    .unwrap();
    assert_eq!(f.g.edge(edge).unwrap().unwrap().invalid_at, Some(3_000));
    // A later timestamp can never re-extend the interval.
    f.g.apply(GraphOp::InvalidateEdge {
        edge_id: edge,
        invalid_at: 6_000,
    })
    .unwrap();
    assert_eq!(f.g.edge(edge).unwrap().unwrap().invalid_at, Some(3_000));
}

/// System-timeline sanity on every row grit writes: expired_at (when set) is
/// never before created_at.
#[test]
fn system_timeline_never_inverts_on_merge() {
    let f = fixture(5_000);
    let a = node(&f.g, "Left");
    let b = node(&f.g, "Right");
    f.clock.set_ms(9_000);
    f.g.apply(GraphOp::MergeNodes { from: b, into: a }).unwrap();

    let merged = f.g.node(b).unwrap().unwrap();
    assert!(merged.expired_at.unwrap() >= merged.created_at);
}

/// Searching with as_of must hide invalidated facts, exactly like traversal.
#[test]
fn search_respects_validity() {
    let f = fixture(1_000);
    let a = node(&f.g, "Kernel");
    let b = node(&f.g, "Image");
    let edge = f.g.new_id();
    f.g.apply(GraphOp::AddEdge {
        id: edge,
        src: a,
        dst: b,
        rel: "EQUALS".into(),
        fact: "the kernel equals the image here".into(),
        attrs: json!({}),
        group_id: String::new(),
        valid_at: Some(1_000),
        invalid_at: None,
    })
    .unwrap();
    f.g.apply(GraphOp::InvalidateEdge {
        edge_id: edge,
        invalid_at: 2_000,
    })
    .unwrap();

    let edge_hits = |q: Query| {
        f.g.search(q.budget(Budget::items(10)))
            .unwrap()
            .into_iter()
            .filter(|h| matches!(h.target, grit_core::SearchTarget::Edge(_)))
            .count()
    };
    assert_eq!(edge_hits(Query::text("kernel image").as_of(1_500)), 1);
    assert_eq!(edge_hits(Query::text("kernel image").as_of(2_500)), 0);
}