Trusty/crates/adapters/trusty-tokio/tests/runtime.rs

//! Does the async facade actually keep the runtime free?
//!
//! That is the only claim it makes, so it is the one worth testing.
//! Everything else here exists because it is a way the facade could be
//! quietly wrong: a lost value, a swallowed panic, a truncation that
//! reads as a complete chain.

use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use std::time::Duration;

use trusty_store::{
    KvCrossCheckReport, LastRow, RowInsert, Store, StoreError, StoreResult, StoredRow, TipRow,
    TipUpsert,
};
use trusty_tokio::{AsyncStore, BlockingStore};

/// A store that is deliberately slow, so blocking is observable.
struct SlowStore {
    delay: Duration,
    rows: usize,
    inserts: Arc<AtomicUsize>,
    panic_on_insert: bool,
}

impl SlowStore {
    fn new(delay_ms: u64) -> Self {
        Self {
            delay: Duration::from_millis(delay_ms),
            rows: 0,
            inserts: Arc::new(AtomicUsize::new(0)),
            panic_on_insert: false,
        }
    }

    fn with_rows(mut self, rows: usize) -> Self {
        self.rows = rows;
        self
    }
}

fn stored_row(sequence: i64) -> StoredRow {
    StoredRow {
        id: sequence + 1,
        entity_id: "e".into(),
        chain_id: "c".into(),
        event_type: "MUTATION".into(),
        sequence,
        prev_hash: String::new(),
        row_hash: Some(format!("{sequence:064}")),
        anchor_hash: None,
        raw_payload: None,
        detected_at: String::new(),
        received_at: String::new(),
        source_url: String::new(),
        extraction_method: "trusty".into(),
    }
}

impl Store for SlowStore {
    fn last_row(&self, _entity_id: &str) -> StoreResult<Option<LastRow>> {
        std::thread::sleep(self.delay);
        Ok(Some(LastRow {
            row_hash: "aa".into(),
            sequence: 7,
        }))
    }

    fn insert_row(&mut self, _row: &RowInsert) -> StoreResult<i64> {
        std::thread::sleep(self.delay);
        if self.panic_on_insert {
            panic!("a bug inside a store operation");
        }
        Ok(self.inserts.fetch_add(1, Ordering::SeqCst) as i64 + 1)
    }

    fn iter_chain<'a>(
        &'a self,
        _entity_id: &str,
    ) -> StoreResult<Box<dyn Iterator<Item = StoreResult<StoredRow>> + 'a>> {
        Ok(Box::new((0..self.rows as i64).map(|i| Ok(stored_row(i)))))
    }

    fn get_tip(&self, _chain_id: &str) -> StoreResult<Option<TipRow>> {
        Ok(None)
    }

    fn get_tip_for_entity(&self, _entity_id: &str) -> StoreResult<Option<TipRow>> {
        Ok(None)
    }

    fn set_tip(&mut self, _tip: &TipUpsert) -> StoreResult<()> {
        Ok(())
    }

    fn kv_cross_check(&self) -> StoreResult<KvCrossCheckReport> {
        Err(StoreError::Backend("not implemented for this test".into()))
    }

    fn with_txn(
        &mut self,
        f: &mut dyn FnMut(&mut dyn Store) -> StoreResult<()>,
    ) -> StoreResult<()> {
        let mut scratch = SlowStore::new(0);
        f(&mut scratch)
    }
}

fn row_insert() -> RowInsert {
    RowInsert {
        entity_id: "e".into(),
        chain_id: "c".into(),
        event_type: "MUTATION".into(),
        sequence: 0,
        prev_hash: String::new(),
        row_hash: None,
        anchor_hash: None,
        raw_payload: None,
        detected_at: String::new(),
        received_at: String::new(),
        source_url: String::new(),
        extraction_method: "trusty".into(),
    }
}

/// **The claim.** A single-worker runtime, one slow store call, and an
/// unrelated task that must keep running while it is in flight.
///
/// Single-threaded on purpose: with more than one worker the test would
/// pass even if the call blocked, because another worker would pick the
/// second task up. One worker is the only configuration where blocking
/// is fatal and therefore the only one that proves anything.
#[tokio::test(flavor = "current_thread")]
async fn a_slow_store_call_does_not_park_the_runtime() {
    let store = BlockingStore::new(SlowStore::new(300));
    let ticks = Arc::new(AtomicUsize::new(0));

    let counter = Arc::clone(&ticks);
    let ticker = tokio::spawn(async move {
        for _ in 0..20 {
            tokio::time::sleep(Duration::from_millis(10)).await;
            counter.fetch_add(1, Ordering::SeqCst);
        }
    });

    store.insert_row(row_insert()).await.expect("insert");
    let progressed = ticks.load(Ordering::SeqCst);
    ticker.abort();

    assert!(
        progressed > 5,
        "the other task must have kept running while the store was busy. \
         It ticked {progressed} times; blocking the worker would have left \
         it at or near zero, which is exactly the runtime starvation this \
         facade exists to prevent"
    );
}

#[tokio::test(flavor = "current_thread")]
async fn concurrent_callers_serialise_rather_than_race() {
    let store = BlockingStore::new(SlowStore::new(20));
    let mut handles = Vec::new();
    for _ in 0..8 {
        let s = store.clone();
        handles.push(tokio::spawn(async move { s.insert_row(row_insert()).await }));
    }
    let mut ids = Vec::new();
    for h in handles {
        ids.push(h.await.expect("join").expect("insert"));
    }
    ids.sort_unstable();

    assert_eq!(
        ids,
        (1..=8).collect::<Vec<i64>>(),
        "every caller must get its own row and none may be lost. The \
         backend serialises writes; that is its property, made visible \
         here rather than hidden"
    );
}

#[tokio::test]
async fn a_panic_in_the_store_becomes_an_error_not_a_dead_task() {
    let mut slow = SlowStore::new(0);
    slow.panic_on_insert = true;
    let store = BlockingStore::new(slow);

    let result = store.insert_row(row_insert()).await;

    assert!(
        matches!(result, Err(StoreError::Backend(_))),
        "a panic inside a store operation is a bug, but one that unwinds \
         into an unrelated task is far harder to find than one arriving as \
         an error at the call site that caused it"
    );
}

#[tokio::test]
async fn a_bounded_read_says_when_it_left_something_behind() {
    let store = BlockingStore::new(SlowStore::new(0).with_rows(10));

    let (rows, more) = store.chain_rows_bounded("e", 4).await.expect("bounded");
    assert_eq!(rows.len(), 4);
    assert!(
        more,
        "a truncated history that looks complete is how a verifier \
         concludes a chain is intact having seen only the part that is"
    );

    let (all, more) = store.chain_rows_bounded("e", 50).await.expect("bounded");
    assert_eq!(all.len(), 10);
    assert!(!more, "with nothing left behind it must not claim there is");
}

#[tokio::test]
async fn the_whole_chain_can_still_be_read_when_that_is_wanted() {
    let store = BlockingStore::new(SlowStore::new(0).with_rows(5));
    let rows = store.chain_rows("e").await.expect("chain");
    assert_eq!(
        rows.len(),
        5,
        "iter_chain cannot cross a thread boundary, so this materialises \
         instead of streaming -- but it must still return everything"
    );
}

#[tokio::test]
async fn a_transaction_carries_its_result_out_by_value() {
    let store = BlockingStore::new(SlowStore::new(0));

    let answer = store
        .with_txn_owned(|txn| {
            txn.insert_row(&row_insert())?;
            Ok("committed".to_string())
        })
        .await
        .expect("txn");

    assert_eq!(
        answer, "committed",
        "the blocking with_txn captures results through outer state \
         because its closure cannot return one; an owned closure can, \
         which removes the chance of reading a captured slot on a path \
         where the transaction rolled back"
    );
}

#[tokio::test]
async fn a_failing_transaction_reports_the_failure_not_a_missing_value() {
    let store = BlockingStore::new(SlowStore::new(0));

    let result: StoreResult<String> = store
        .with_txn_owned(|_txn| Err(StoreError::Backend("deliberate".into())))
        .await;

    match result {
        Err(StoreError::Backend(msg)) => assert_eq!(
            msg, "deliberate",
            "the body's own error must survive, not be replaced by a \
             complaint that no value was produced"
        ),
        other => panic!("expected the body's error, got {other:?}"),
    }
}

#[tokio::test]
async fn clones_share_one_store() {
    let store = BlockingStore::new(SlowStore::new(0));
    let clone = store.clone();

    store.insert_row(row_insert()).await.expect("first");
    let second = clone.insert_row(row_insert()).await.expect("second");

    assert_eq!(
        second, 2,
        "a clone must talk to the same store, or two halves of an \
         application would be appending to different chains while \
         believing they shared one"
    );
}