nordisk/nornir

crates/mimir-room/tests/room/live.rs

//! A room opened WHILE the consumer writes the same log (wiring map R13), a
//! room from a sync caller (R14), and the typed "never consumed" (R15).
//!
//! The consumer is held inside a commit by a gated dumper — where a real
//! consume spends its minutes (`cargo check`) — on its own thread; rooms are
//! opened from the test thread over a clone of the same log handle.

use std::path::Path;
use std::sync::mpsc::{Receiver, Sender, channel};
use std::time::Duration;

use anyhow::Result;
use mimir_ingest::{CommitDump, Consumer, Dumper};
use mimir_room::{Room, Scope, ScopeError};
use mimir_schema::EdgeKind;

use super::common::*;

const WAIT: Duration = Duration::from_secs(60);

/// The scripted dumper, stopping inside its `hold_at`-th dump until released.
struct Gated {
    inner: Scripted,
    calls: usize,
    hold_at: usize,
    entered: Sender<()>,
    go: Receiver<()>,
}

impl Dumper for Gated {
    fn dump(&mut self, worktree: &Path, sha: &str) -> Result<CommitDump> {
        self.calls += 1;
        if self.calls == self.hold_at {
            self.entered.send(()).unwrap();
            self.go.recv_timeout(WAIT).expect("the test never released the gated dump");
        }
        self.inner.dump(worktree, sha)
    }
}

/// `Room::open_blocking` on a thread of its own, so a blocked open is a test
/// failure with a reason instead of a hung test.
fn open_within(log: &mimir_ingest::EventLog, scope: Scope, what: &str) -> mimir_room::Opened {
    let (tx, rx) = channel();
    let log = log.clone();
    std::thread::spawn(move || tx.send(Room::open_blocking(&log, &scope)).ok());
    rx.recv_timeout(Duration::from_secs(20))
        .unwrap_or_else(|_| panic!("{what}: the room did not open within 20 s — a reader is waiting for the writer"))
        .unwrap_or_else(|e| panic!("{what}: {e:#}"))
}

#[test]
fn a_room_opened_during_an_in_flight_consume_is_the_last_committed_release() {
    let rt = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap();
    let _g = rt.block_on(serial());
    Room::clear_cache();
    let tmp = tempfile::tempdir().unwrap();
    let log = rt.block_on(open_log(tmp.path()));
    let reader = log.clone();
    let steps = room_steps();
    assert_eq!(steps.len(), 3);
    scratch(&tmp.path().join(format!("src-{ROOM}")), steps.len());
    let spec = repo_spec(tmp.path(), ROOM);
    let (entered_tx, entered) = channel();
    let (go, go_rx) = channel();

    // the consumer: held inside the dump of commit 3
    let writer = std::thread::spawn(move || {
        tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap().block_on(async {
            let gated = Gated { inner: Scripted(steps), calls: 0, hold_at: 3, entered: entered_tx, go: go_rx };
            let mut c = Consumer::boot(log, gated).await.unwrap();
            c.run(&spec).await.unwrap().len()
        })
    });
    entered.recv_timeout(WAIT).expect("the consumer never reached commit 3");
    assert!(reader.has_writer(), "the consume is in flight");

    // ---- the room opens NOW, and it is release 2: e→a is in, c→x(ffi) is out
    let during = open_within(&reader, Scope::current(), "during the consume");
    assert!(!during.cached);
    assert_eq!(during.room.scope().repos[ROOM], 2, "the last committed release, not the one in flight");
    let callers = during.room.dependents_by(&["k a"], &[EdgeKind::Calls], None, None).unwrap();
    assert!(during.room.monikers(&callers.value.reached).contains(&"k e"), "seq 2 added e→a");
    assert!(during.room.ordinal("k d").is_some(), "seq 3 (in flight) removes d: not yet");
    // a second reader at the same time gets the same snapshot from the cache
    let second = open_within(&reader, Scope::current(), "a second reader during the consume");
    assert!(second.cached && second.room.same_snapshot(&during.room));

    go.send(()).unwrap();
    assert_eq!(writer.join().unwrap(), 3);

    // ---- the next open is release 3; the room held from before did not move
    let after = open_within(&reader, Scope::current(), "after the consume");
    assert!(!after.cached && !after.room.same_snapshot(&during.room));
    assert_eq!(after.room.scope().repos[ROOM], 3);
    assert!(after.room.ordinal("k d").is_none(), "seq 3 removed d");
    assert_eq!(during.room.scope().repos[ROOM], 2);
    assert!(during.room.ordinal("k d").is_some(), "a room is an immutable snapshot");
}

#[test]
fn a_repo_that_was_never_consumed_is_a_typed_scope_error() {
    let rt = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap();
    let _g = rt.block_on(serial());
    let tmp = tempfile::tempdir().unwrap();
    let log = rt.block_on(fixture(tmp.path()));

    // sync entry (R14), from a thread with no runtime
    let room = Room::at_blocking(&log, &Scope::repo(PLAIN, None)).unwrap();
    assert_eq!(room.scope().repos[PLAIN], 1);
    // … and from inside a runtime's blocking pool
    let l2 = log.clone();
    let n = rt.block_on(async move { tokio::task::spawn_blocking(move || Room::at_blocking(&l2, &Scope::current()).map(|r| r.len())).await.unwrap() }).unwrap();
    assert!(n > 0);

    // R15: typed, not a string to grep
    let err = Room::at_blocking(&log, &Scope::repo("nobody", None)).err().expect("never consumed");
    match err.downcast_ref::<ScopeError>() {
        Some(ScopeError::NotConsumed { repo, known }) => {
            assert_eq!(repo, "nobody");
            assert_eq!(known, &vec![PLAIN.to_string(), ROOM.to_string()]);
        }
        other => panic!("expected ScopeError::NotConsumed, got {other:?} ({err:#})"),
    }
    let past = Room::at_blocking(&log, &Scope::repo(PLAIN, Some(9))).err().unwrap();
    assert!(matches!(past.downcast_ref::<ScopeError>(), Some(ScopeError::PastOffset { seq: 9, offset: 1, .. })), "{past:#}");
}